本文目录一览：
- 1、给kafka配置外部连接
- 2、kafka集群测试正常，但是Java连接kafka出现异常，急求大神解答！！！！！！！！！！！
- 3、Kafka（四）集群之kafka
- 4、kafka集群配置和使用
- 5、服务端技术实战系列——Kafka篇
- 6、Docker搭建Kafka测试集群
给kafka配置外部连接
在公司的测试环境中，有的应用需要远程连接kafka，本地有时也是需要连接到kafka进行Debug，这就需要将kafka配置成外部可连接。想要实现这种效果，有两种实现方法，第一种方法是将所有kafka的连接都配置成公网IP连接。第二种方法是采用kafka的内外分离配置。方法一虽然能够实现kafka的外部连接，可是服务器上面对kafka的连接也会默认使用公网IP的方式连接，而不是内网，这样会给实例的公网带宽带来很大的压力，应用一多，就会造成实例无法进行登录。而方法二就可以有效的避免这种现象了，它是将云上本地应用采用内网来连接kafka，而同时又采用不同的端口配置外网连接，这样能够有效的减少公网带宽的压力。
在原有配置的基础上加上或者更改如下配置,配置接受外网连接的端口为9093，同时打开安全组的9093端口，配置过后通过重启kafka即可通过9093在本地连接kafka。
listener.security.protocol.map=INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT主要分别定义内部和外部连接采用的安全协议
listeners=INTERNAL://内网IP:9092,EXTERNAL://0.0.0.0:9093主要是定义内部和外部连接监听的地址端口
advertised.listeners=INTERNAL://内网IP:9092,EXTERNAL://外网IP:9093主要是提交给zookeeper来实现对kafka内部和外部的连接，最开始改配置只是配置了外部的连接，没有内部的连接，所以对kafka的连接都是通过外部连接。
inter.broker.listener.name=INTERNAL主要是制定kafka集群内部broker之前通过INTERNAL的配置来进行内部通讯。
参考连接：
kafka集群测试正常，但是Java连接kafka出现异常，急求大神解答！！！！！！！！！！！
首先你在链接时候检查是否代码里的IP 和端口是不是对的，端口是broker 端口，默认9092 ；
其次查看代码是生产者，看Kafka 集群里这个主题是否存在（如果不存在，默认是配置可以自动创建，看是非将该配置修改）；然后检测防火墙，相应端口是否开放（防火墙直接关也可以）；检测 server.properties 文件的 listeners 是否配置，若没有将其配置好
Kafka（四）集群之kafka
在章节二（ ）中，我们部署了单机的kafka，现在我们部署一套集群模式的kafka。
这里我准备了三台虚拟机：
192.168.184.134
192.168.184.135
192.168.184.136
每台机器部署一个zk和kafka。
上一章节中zk集群已经部署完毕。
在章节二中，134这台机器已经有kafka存在了，我们在另外两台机器上安装kafka：
在上面的文件中有几个关键点，我们一一进行配置，我会对配置中的说明翻译：
以下这两个listeners，advertised_listeners 是对外暴露的服务端口，真正建立连接用的是 listeners。
在内网中我们使用listenners就可以了，在docker等容器或云中使用advertised。
下面这个是日志路径的配置
下面这个是个重点的东西，topic在磁盘上会分为多个partitions存储，相比单一文件存储，增加了并行性，在后续文章中会详细去讲解：
日志的保存时间：
以下是zookeeper的配置：
这里我们直接设置后台启动，三个节点都是如此：
这里面有个小坑，还记得之前我们搭建的单机环境吗？那时候默认的日志文件夹在/tmp/kafka-logs下面，生成了很多内容，导致我们134这个节点无法启动成功，报错如下：
解决这个问题只需要把/tmp/kafka-logs文件删除就好了。
看到日志出现这一句表明启动成功了：
下面我们验证下是否搭建成功了，首先使用kafkatool工机具连接看下：
我们在134节点创建一个topic：
查看topic列表：
在kafkatool中查看：
创建生产者：
创建消费者：
生成者发送消息：
消费者接收消息：
到此为止，kafka的集群搭建已经完成了。在后面的文章我们会去学习如何在springboot中集成kafka。
kafka集群配置和使用
进入安装目录，修改server.properties文件
修改如下属性，除id外，其他每台主机一致：
语义配置：（可选）
先启动zookeeper集群，已经在三台主机上配置好了zookeeper集群，启动：
在各台主机上进入zookeeper目录，分别启动zk：
在各台主机上进入kafka目录，分别启动kafka：
启动结果为：
kafka占据了前台，要使用主机，需要打开新终端
在新打开的终端上，进入zk目录，
进入kafka目录，创建主体
服务端技术实战系列——Kafka篇
一．概念原理
[if !supportLists]1. [endif]主题(topic)：主题是对消息的分类。
[if !supportLists]2. [endif]消息(message)：消息是kafka通信的基本单位。
[if !supportLists]3. [endif]分区(partition)： 一组 消息对应 一个 主题， 一个 主题对应 一个或多个 分区。每个分区为一系列有序消息组成的 有序队列 ；每个分区在物理上对应一个文件夹。
[if !supportLists]4. [endif]副本(replica)：每个分区有 一个或多个 副本，分区的副本分布在集群的 不同 代理（机器）上，以提高可用性；分区的副本与日志对象是一一对应的。
[if !supportLists]5. [endif]Kafka只保证一个 分区内 的消息 有序性 ，不保证跨分区消息的有序性。消息被追加到相应分区中， 顺序写入磁盘 ，效率非常高。
[if !supportLists]6. [endif]Kafka选取某个某个分区的 一个 副本作为leader副本，该分区的 其他 副本为follower副本。 只有leader副本负责处理客户端读/写请求 ，follower副本从leader副本同步数据。
[if !supportLists]7. [endif]任何发布到分区的消息都会追加到日志文件的尾部， 每条消息 在日志文件中的 位置 都对应一个 按序递增的偏移量 ；偏移量在一个分区下严格有序。
[if !supportLists]8. [endif]Kafka不允许对消息进行随机读写。
[if !supportLists]9. [endif]新版消费者将 消费偏移量 保存到kafka内部的一个主题中。
[if !supportLists]10. [endif]Kafka集群由 一个或多个代理 （Broker，也称为kafka实例）构成。可以在 一台 服务器上配置 一个或多个代理 ，每个代理具有唯一标识broker.id。
[if !supportLists]11. [endif]生产者将消息 发送给代理 （Broker）。
[if !supportLists]12. [endif]消费者以 拉取 （pull）方式拉取数据，每个消费者都属于一个消费组。
[if !supportLists]13. [endif]同一个主题的一条消息只能被 同一个消费组 下的某一个消费者消费，但 不同消费组 的消费者可以 同时 消费该消息。
[if !supportLists]14. [endif]消息 广播 ：指定各消费者属于不同消费组；消息 单播 ：指定各消费者属于同一个消费组。
[if !supportLists]15. [endif]Kafka启动时在Zookeeper上创建相应节点来保存 元数据 ，元数据包括：代理节点信息、集群信息、主题信息、分区状态信息、分区副本分配方案、动态配置等；
[if !supportLists]16. [endif]Kafka通过 监听 机制在节点注册监听器来监听节点元数据变化；
[if !supportLists]17. [endif]Kafka将数据写入 磁盘 ，以文件系统来存数据；
[if !supportLists]18. [endif]生产环境一般将zookeeper集群和kafka集群 分机架 部署；
[if !supportLists]二．[endif] Kafka Producer
配置：
/**
* xTestProxy——KafkaConfigConstant
*
* @author ZhangChi
* @date 2018年6月20日---下午5:50:44
* @version 1.0
*/
public class KafkaConfigConstant {
public static final String KAFKA_CLUSTER = "fa-common1.hangzhou-1.kafka.internal.lede.com:9200,fa-common2.hangzhou-1.kafka.internal.lede.com:9200,fa-common3.hangzhou-1.kafka.internal.lede.com:9200";
}
生产者配置：
/**
* xTestProxy——HttpKafkaProducerFactory
*
* @author ZhangChi
* @date 2018年6月11日---下午2:37:51
* @version 1.0
*/
public class HttpKafkaProducerFactory {
// 真正的KafkaProducer仅有一份
private static KafkaProducer kafkaProducer = null ;
private static Properties property ;
public static KafkaProducer getKafkaProducer() {
if ( kafkaProducer == null ) {
synchronized (HttpKafkaProducerFactory. class ) {
if ( kafkaProducer == null ) {
property = buildKafkaProperty ();
kafkaProducer = new KafkaProducer( property );
}
}
}
return kafkaProducer ;
}
public static Properties buildKafkaProperty() {
Properties props = new Properties();
props.put(ProducerConfig. BOOTSTRAP_SERVERS_CONFIG , KafkaConfigConstant. KAFKA_CLUSTER );
props.put(ProducerConfig. ACKS_CONFIG , "all");
props.put(ProducerConfig. RETRIES_CONFIG , 0);
props.put(ProducerConfig. BATCH_SIZE_CONFIG , 16384);
props.put(ProducerConfig. BUFFER_MEMORY_CONFIG , 33554432);
props.put(ProducerConfig. LINGER_MS_CONFIG , 1);
props.put(ProducerConfig. KEY_SERIALIZER_CLASS_CONFIG , "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig. VALUE_SERIALIZER_CLASS_CONFIG ,
"org.apache.kafka.common.serialization.StringSerializer");
return props;
}
}
生产者线程组：
/**
* xTestProxy——HttpKafkaProducerThread
* 多线程每次new一个实例
*
* @author ZhangChi
* @date 2018年6月25日---下午2:09:39
* @version 1.0
*/
public class HttpKafkaProducerThread implements Runnable {
private static Logger logger = LoggerFactory. getLogger ("HttpKafkaProducerThread");
private final String KAFKA_TOPIC = KafkaConstant. HTTP_REQ_RESP_TOPIC ;
private String kafkaMessageJson;
private KafkaProducer producer;
public String messageType;
public String originalMessage;
private static KafkaMessage kafkaMessage = new KafkaMessage();
public HttpKafkaProducerThread(KafkaProducer producer, String messageType, String originalMessage) {
this .producer = producer;
this .messageType = messageType;
this .originalMessage = originalMessage;
}
@Override
public void run() {
// TODO Auto-generated method stub
/* 1.构建kafka消息*/
kafkaMessageJson = generateKafkaMessage( this .messageType, this .originalMessage);
/* 2.发送kafka消息*/
if (kafkaMessageJson != null !StringUtils. isEmpty (kafkaMessageJson)) {
logger .info("create message start:" + kafkaMessageJson);
producer.send( new ProducerRecord( this .KAFKA_TOPIC, kafkaMessageJson));
} else {
logger .info("kafkaMessageJson is null!");
}
}
private String generateKafkaMessage(String messageType, String originalMessage) {
if (StringUtils. isBlank (messageType) || StringUtils. isBlank (originalMessage)) {
return null ;
}
kafkaMessage .setMessageId(KafkaMessageUtils. generateId ());
kafkaMessage .setMessageTime(KafkaMessageUtils. generateTime ());
kafkaMessage .setMessageType(messageType);
kafkaMessage .setMessage(originalMessage);
String kafkaMessageToJson = null ;
try {
kafkaMessageToJson = KafkaMessageUtils. objectToJson ( kafkaMessage );
} catch (JsonProcessingException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
kafkaMessageJson = kafkaMessageToJson;
return kafkaMessageToJson;
}
}
[if !supportLists]三．[endif] Kafka Consumer
消费者配置：
private static Properties buildKafkaProperty() {
Properties properties = new Properties();
// 测试环境kafka的端口号是9200
properties.put(ConsumerConfig. BOOTSTRAP_SERVERS_CONFIG , KafkaConfigConstant. KAFKA_CLUSTER );
// 消费组名称
properties.put(ConsumerConfig. GROUP_ID_CONFIG , KafkaConfigConstant. GROUP_ID );
properties.put(ConsumerConfig. CLIENT_ID_CONFIG , "test");
// 从头消费
properties.put(ConsumerConfig. AUTO_OFFSET_RESET_CONFIG , "earliest");
// 自动提交偏移量
properties.put(ConsumerConfig. ENABLE_AUTO_COMMIT_CONFIG , "true");
// 时间间隔1s
properties.put(ConsumerConfig. AUTO_COMMIT_INTERVAL_MS_CONFIG , "1000");
properties.put(ConsumerConfig. KEY_DESERIALIZER_CLASS_CONFIG ,
"org.apache.kafka.common.serialization.StringDeserializer");
properties.put(ConsumerConfig. VALUE_DESERIALIZER_CLASS_CONFIG ,
"org.apache.kafka.common.serialization.StringDeserializer");
return properties;
}
消费者线程组：
/**
* AnalysisEngine——HttpKafkaConsumerGroup
*
* @author ZhangChi
* @date 2018年6月11日---下午6:20:47
* @version 1.0
*/
@Service("httpKafkaConsumerGroup")
public class HttpKafkaConsumerGroup {
@Autowired
private RequestAnalyzer requestAnalyzer;
@Autowired
private EsDocumentServiceImpl esDocumentServiceImpl;
@Autowired
private AnalysisEngineClient analysisEngineClient;
@Autowired
private MongoTemplate mongoTemplate;
private List httpKafkaConsumerList = new ArrayList();
public void initHttpKafkaConsumerGroup( int consumerNumber, RunModeEnum mode) {
for ( int i = 0; i consumerNumber; i++) {
/**
* 将注入的服务当做构造参数，这样保证每个子线程都能拿到服务实例而不是空指针！
*/
HttpKafkaConsumer consumerThread = new HttpKafkaConsumer(requestAnalyzer, esDocumentServiceImpl, mode, analysisEngineClient, mongoTemplate);
httpKafkaConsumerList.add(consumerThread);
}
}
public void consumeGroupStart() {
for (HttpKafkaConsumer item : httpKafkaConsumerList) {
LogConstant. runLog .info("httpKafkaConsumerList size : " + httpKafkaConsumerList.size());
Thread consumerThread = new Thread(item);
consumerThread.start();
}
}
}
先逐个初始化消费者实例，然后将这些消费者加入到消费组列表中。消费组启动后，会循环产生消费者线程。
Docker搭建Kafka测试集群
zookeeper利用这个hostname在集群中的broker之间同步消息, 这里配置为kafka在docker虚拟网络中的IP。 使用你的宿主机的IP也可以，但是端口号要改成宿主机的端口号
Kafka在Docker container中的端口号
如果你需要在Docker虚拟网络之外使用Kafka集群，你需要把这两个参数配置成你宿主机的IP，端口号要改成Docker映射到宿主机的端口号（9092， 9093）。当往Kafka其中的一个broker发送消息时，Kafka集群从ZooKeeper取得Broker IP和端口号，然后同步数据，使用虚拟网络IP会导致发送和接收消息失败，因为宿主机无法访问Docker虚拟网络内的节点
例如我的宿主机IP是，192.168，0.2，
上述2个Kafka容器，9092端口号分别映射到宿主机的9092和9093端口。
环境变量配置如下
Kafka1配置为
KAFKA_ADVERTISED_HOST_NAME: 192.168.0.2
KAFKA_ADVERTISED_PORT: 9092
Kafka2配置为
KAFKA_ADVERTISED_HOST_NAME: 192.168.0.2
KAFKA_ADVERTISED_PORT: 9093
