SpringBoot整合kafka

猿码优创

发布于 2019-07-27 20:11:23

1K0

发布于 2019-07-27 20:11:23

经过前三篇文章安装jdk 安装zookeeper 以及安装kafka 全部已经竣工了，不知道小伙伴们成功搭建kafka了不。
憋了三天的大招，今天放出来吧。今天大家用java代码连接kafka。
第一步：修改kafka的server.properties文件

命令： 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

重启防火墙

  重启防火墙：service iptables restart

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

我们来写一下消息生成者创建一个SpringBoot kafka_product Demo
目录结构如下：

	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 #选择要用那个配置文件

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

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

测试结果：

我们再来封装一下消费者（可以直接在生产者项目写消费者信息，但是为了给你们展示清楚，我就分成两个项目了。）：
创建一个SpringBoot kafka_consumer Demo
目录结构：

代码如下：
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>

(adsbygoogle =window.adsbygoogle ||[]).push({});

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和端口

启动项目：

启动命令行生产者

测试结果：

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

本文参与腾讯云自媒体分享计划，分享自作者个人站点/博客。

原始发表：2019-06-14，如有侵权请联系 cloudcommunity@tencent.com 删除

kafka

网络安全

本文分享自作者个人站点/博客前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体分享计划，欢迎热爱写作的你一起参与！

kafka

网络安全

登录后参与评论

0 条评论

热度

SpringBoot整合kafka

SpringBoot整合kafka

社区

活动

资源

关于

腾讯云开发者

热门产品

热门推荐

更多推荐