经过前三篇文章 安装jdk 安装zookeeper 以及安装kafka 全部已经竣工了,不知道小伙伴们成功搭建kafka了不。

憋了三天的大招,今天放出来吧。今天大家用java代码连接kafka。

第一步:修改kafka的server.properties文件

file

命令: vi server.properties 
修改内容:
broker.id=0
port=9092
host.name=192.168.241.134
log.dirs=/DATA/kafka/kafka_2.12-2.0.0/log
zookeeper.connect=localhost:2181

第二步开发端口9092端口

1.可以直接关闭防火墙、

centos6关闭防火墙命令以下:
	1. 永久性生效
	开启:chkconfig iptables on
	关闭:chkconfig iptables off
	
	2. 即时生效,重启后失效
	开启:service iptables start
	关闭:service iptables stop
	
CentOS 7.0默认使用的是firewall作为防火墙,使用iptables必须重新设置一下
	1、直接关闭防火墙
	systemctl stop firewalld.service #停止firewall
	systemctl disable firewalld.service #禁止firewall开机启动

2、修改iptables 文件

centos 6修改方法:
执行命令:vi /etc/sysconfig/iptables
然后在文件中增加一行
-A RH-Firewall-1-INPUT -m state –state NEW -m tcp -p tcp –dport 9092 -j ACCEPT

file

centos7 设置 iptables service
yum -y install iptables-services
如果要修改防火墙配置, 增加kafka端口9092
vi /etc/sysconfig/iptables 
增加规则
-A INPUT -m state --state NEW -m tcp -p tcp --dport 9092 -j ACCEPT
保存退出后
systemctl restart iptables.service #重启防火墙使配置生效
systemctl enable iptables.service #设置防火墙开机启动

启动kafka

1、启动zookeeper
命令: sh $zookeeper_home/bin/zkServer.sh start
2、启动kafka
命令:在kafka目录下 输入./kafkaStart.sh

file

我们来写一下消息生成者 创建一个SpringBoot kafka_product Demo

目录结构如下:

file

	pom代码:
	<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>cn.cnbuilder</groupId>
    <artifactId>kafka</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>kafka</name>
    <description>kafka Demo project for Spring Boot</description>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>1.5.8.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <java.version>1.8</java.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-freemarker</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <version>1.0.6.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>

KafkaProducer代码如下:

package cn.cnbuilder.kafka.producer;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;

import java.util.UUID;

/**
 * 生产者
 * 使用@EnableScheduling注解开启定时任务
 */
@Component
@EnableScheduling
public class KafkaProducer {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    /**
     * 定时任务1
     */
    @Scheduled(cron = "00/1 * * * * ?")
    public void send(){
        String message = UUID.randomUUID().toString();
        ListenableFuture future = kafkaTemplate.send("test", message);
        future.addCallback(o -> System.out.println("消息发送成功:" + message), throwable -> System.out.println("消息发送失败:" + message));
    }
}

KafkaApplication代码:

 package cn.cnbuilder.kafka;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class KafkaApplication {

    public static void main(String[] args) {
        SpringApplication.run(KafkaApplication.class, args);
    }

}

application-dev.yml代码:

server:
  port: 8888
spring:
  kafka:
    producer:
      bootstrap-servers: 192.168.241.134:9092 #服务器ip+端口

application.yml代码:

spring:
  profiles:
    active: dev #选择要用那个配置文件

项目构建完成我们启动一下:

file

然后在服务器启动一下消费者

file

测试结果:

file

我们再来封装一下消费者(可以直接在生产者项目写消费者信息,但是为了给你们展示清楚,我就分成两个项目了。):

创建一个SpringBoot kafka_consumer Demo

目录结构:

file

代码如下:

pom文件代码:

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>cn.cnbuilder</groupId>
    <artifactId>consumer</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>kafka_consumer</name>
    <description>kafka_consumer Demo project for Spring Boot</description>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>1.5.8.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <java.version>1.8</java.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-freemarker</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <version>1.0.6.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>

KafkaConsumer代码:

package cn.cnbuilder.consumer.consumer;

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/**
 * 消费者
 * 使用@KafkaListener注解,可以指定:主题,分区,消费组
 */
@Component
public class KafkaConsumer {

    @KafkaListener(topics = {"test"})
    public void receive(String message){
        System.out.println("test--消费消息:" + message);
    }
}

KafkaConsumerApplication

package cn.cnbuilder.consumer;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class KafkaConsumerApplication {

    public static void main(String[] args) {
        SpringApplication.run(KafkaConsumerApplication.class, args);
    }

}

application.yml

server:
  port: 8082
spring:
  kafka:
    consumer:
      group-id: test
      bootstrap-servers: 192.168.241.134:9092 # ip和端口

启动项目:

file

启动命令行生产者

file

测试结果:

file

我们现在用代码来生产消息和消费消息(启动两个项目用两个端口号哦!)

file

终、、


以上就是SpringBoot整合kafka,有什么问题可以联系我哈。

百度云盘下载地址:链接: https://pan.baidu.com/s/1gz68Smg8JpsIpBmOfMN1GA 提取码: 8mqd

鼓励作者写出更好的技术文档,就请我喝一瓶哇哈哈哈哈哈哈哈。。你们的赞助决定我更新的速度哦!

微信:

支付宝:


感谢一路支持我的人。。。。。

Love me and hold me
QQ:69673804(16年老号)
EMAIL:itw@tom.com
友链交换
如果有兴趣和本博客交换友链的话,请按照下面的格式在评论区进行评论,我会尽快添加上你的链接。

网站名称:KingYiFan’S Blog
网站地址:http://blog.cnbuilder.cn
网站描述:年少是你未醒的梦话,风华是燃烬的彼岸花。
网站Logo/头像: [头像地址](https://blog.cnbuilder.cn/upload/2018/7/avatar20180720144536200.jpg)