springboot整合kafka

时间 2019-11-11

标签 springboot 整合 kafka 栏目 Spring 繁體版

原文原文链接

为了方便起见安装zk集群和kafka集群我就不详细说明了，如下摘自https://www.cnblogs.com/lentoo/p/7785004.html，文章部份内容我作了修改，由于版本或者命令改变

kafka集群搭建（windows环境下）

1、简介

Kafka 是一个实现了分布式的、具备分区、以及复制的日志的一个服务。它经过一套独特的设计提供了消息系统中间件的功能。它是一种发布订阅功能的消息系统。html

一、名词介绍

Messagejava

消息，就是要发送的内容，通常包装成一个消息对象。spring

Topic数据库

通俗来说的话，就是放置“消息”的地方，也就是说消息投递的一个容器。假如把消息看做是信封的话，那么 Topic 就是一个邮箱 apache

Partition && Logbootstrap

Partition 分区，能够理解为一个逻辑上的分区，像是咱们电脑的磁盘 C:, D:, E: 盘同样,windows

Kafka 为每一个分区维护着一份日志Log文件。服务器

Producers（生产者）app

和其余消息队列同样，生产者一般都是消息的产生方。负载均衡

在 Kafka 中它决定消息发送到指定Topic的哪一个分区上。

Consumers（消费者）

消费者就是消息的使用者，在消费者端也有几个名词须要区分一下。

通常消息队列有两种模式的消费方式，分别是 队列模式 和 订阅模式。

队列模式：一对一，就是一个消息只能被一个消费者消费，不能重复消费。通常状况队列支持存在多个消费者，可是对于一个消息，只会有一个消费者能够消费它。

订阅模式：一对多，一个消息可能被屡次消费，消息生产者将消息发布到Topic中，只要是订阅改Topic的消费者均可以消费。

2、安装zookeeper

一、简介

Kafka使用zookeeper做为其分布式协调框架，很好的将消息生产、消息存储、消息消费的过程结合在一块儿。同时借助zookeeper，kafka可以生产者、消费者和broker在内的因此组件在无状态的状况下，创建起生产者和消费者的订阅关系，并实现生产者与消费者的负载均衡。

二、下载zookeeper

能够到zookeeper官网下载

http://zookeeper.apache.org/releases.html

三、配置zookeeper

（1）下载解压完成后，来到conf文件夹下，有一个 zoo_sample.cfg 官方默认的配置文件。复制一份，重命名为 zoo.cfg

（2）配置，打开zoo.cfg 修改配置信息

#存储内存中数据库快照的位置，若是不设置参数，更新事务日志将被存储到默认位置。

dataDir=../zkData

#日志文件的位置

dataLogDir=../zkLog

#监听端口

clientPort=2181

（3）集群配置

server.1=127.0.0.1:12888:1388

server.2=127.0.0.1:12889:1389

server.3=127.0.0.1:12887:1387

格式： server.A = B:C:D

A:是一个数字，表示第几号服务器

B:服务器IP地址

C:是一个端口号，用来集群成员的信息交换,表示这个服务器与集群中的leader服务器交换信息的端口

D:是在leader挂掉时专门用来进行选举leader所用的端口

完整的配置文件以下

复制两份zookeeper解压好配置后的文件夹，命名为

在对应的文件下下面修改zoo.cfg的监听端口地址

好比：

第一个zookeeper-3.4.6程序修改zoo.cfg 配置文件

clientPort=2181

第二个zookeeper-3.4.6-2程序修改zoo.cfg 配置文件

clientPort=2182

第三个zookeeper-3.4.6-2程序修改zoo.cfg 配置文件

clientPort=2183

建立ServerID

在配置的dataDir目录下面新建一个 myid 文件，文件内容就是对应的id号，

好比：

zookeeper-3.4.6程序 myid 文件的内容为 1

zookeeper-3.4.6-2程序 myid 文件的内容为 2

zookeeper-3.4.6-3程序 myid 文件的内容为 3

我这边配置的目录是

启动zookeeper

在对应的bin目录下启动

zkServer.cmd

以上zk的集群搭建没有问题，下面会有部份内容作修改

3、安装kafka

（1）下载

去官网 http://kafka.apache.org/下载便可这边下载的是

https://www.apache.org/dyn/closer.cgi?path=/kafka/0.8.2.2/kafka_2.9.2-0.8.2.2.tgz

改成：

http://archive.apache.org/dist/kafka/2.0.0/kafka_2.12-2.0.0.tgz

（2）配置

解压后到config文件夹下打开server.properties配置文件进行配置

（3）配置内容

修改或新增如下配置信息

#惟一标识

broker.id=0

#监听端口

port=9092

host.name=127.0.0.1

#消息最大大小

message.max.bytes=50485760

#配置副本数量

default.replication.factor=2

#获取的最大大小

replica.fetch.max.bytes=50485760

#队列中消息持久化存放的位置，能够多个目录，用逗号分开

log.dirs=/tmp/kafka-logs

#默认的分区数

num.partitions=2

#对应着刚刚配置的zookeeper的三个ip与端口地址

zookeeper.connect=127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183

（4）集群配置

复制两份解压后的文件，命名以下

修改部分配置信息

在对应的server.properties中修改

#惟一标识

broker.id=0

broker.id=1

broker.id=2

#监听端口

port=9092

port=9093

port=9094

启动对应的kafka

进入到bin/windows目录下启动kafka并指定配置文件

kafka-server-start.bat ../../config/server.properties

启动过程当中若是遇到Kafka中错误：

Unrecognized VM option ‘UseCompressedOops’ Error: Clould not create the Java Vritual Machine. Error: A fatal exception has occurres . Program will exit.

解决方案：

找到bin/windows/kafka-run-class.bat 文件,

找到112行左右

IF ["%KAFKA_JVM_PERFORMANCE_OPTS%"] EQU [""] (

  set KAFKA_JVM_PERFORMANCE_OPTS=-server -XX:+UseCompressedOops -XX:+UseParNewGC -XX:+UseConcMarkSweepGC -XX:+CMSClassUnloadingEnabled -XX:+CMSScavengeBeforeRemark -XX:+DisableExplicitGC -Djava.awt.headless=true

)

删除掉 -XX:+UseCompressedOops 便可

测试集群

（1）建立一个 topic

kafka-topics.bat --create --zookeeper 127.0.0.1:2181 --replication-factor 1 --partitions 1 --topic test

（2）查看是否建立成功

kafka-topics.bat --list --zookeeper localhost:2181

（3）发送消息

kafka-console-producer.bat --broker-list localhost:9092 --topic test

This is a message

（4）接收消息

kafka-console-consumer.bat --zookeeper localhost:2181 --topic test --from-beginning

改成：

kafka-console-consumer.bat --
bootstrap-server 127.0.0.1:9092, 127.0.0.1:9093, 127.0.0.1:9094 --topic test --f
rom-beginning

不一样客户端能接收到消息，说明配置成功

接下来为正式整合：

生产者：

引入jar包

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

增长配置

spring:
  kafka:
      bootstrapServers: 127.0.0.1:9092, 127.0.0.1:9093, 127.0.0.1:9094
      producer:
        batchSize: 10
        compressionType: snappy
        acks: all

测试类，ps:上述文章中已经建立TOPIC。。。。test

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

@RunWith(SpringRunner.class)
@SpringBootTest
public class KafkaApplicationTests {
	@Autowired
	private KafkaTemplate kafkaTemplate;
	@Test
	public void contextLoads() {
		try {
			ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send("test","luoye", "hello kafka");
			future.addCallback(new ListenableFutureCallback<SendResult< String, String >>() {
				@Override
				public void onSuccess(SendResult<String, String> result) {
					System.out.println(result.toString());
					System.out.println("推送消息成功");
				}

				@Override
				public void onFailure(Throwable throwable) {
					System.out.println("推送消息失败");
				}
			});
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

}

消费者：

引入jar包

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

增长配置类

spring:
    kafka:
      consumer:
        group-id: defaultGroup
      bootstrap-servers: 127.0.0.1:9092, 127.0.0.1:9093, 127.0.0.1:9094

建立监听器

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class Listener {
  @KafkaListener(topics = {"test"})
  public void listen(ConsumerRecord<?, ?> record) {
     System.out.println(record.key());
     System.out.println(record.value());
  }
}