kafka 支持发布订阅

概述

一般消息队列的是实现是支持两种模式的，即点对点，还有一种是topic发布订阅者模式，比如ACTIVEMQ。KAFKA也支持这两种模式，但是实现的原理不一样。

KAFKA 的消息被读取后，并不是马上删除，这样就可以重复读取。kafka 正式利用这种特性实现发布订阅者模式。

即在发布消息的时候，发布一个topic，可以使用配置多个消费者来消费，消费者使用分组来实现。比如一个topic ，有两个分组的消费者订阅。

那么发布一个消息的时候，两个分组的消费者可以读取到此条消息。

实现

配置两组消费者。

分组1

 <bean id="consumerProperties" class="java.util.HashMap"><constructor-arg><map><!-- 配置kafka的broke --><entry key="bootstrap.servers" value="${kafka.brokerurl}"/><!-- 配置组--><entry key="group.id" value="group1"/><entry key="enable.auto.commit" value="true"/><entry key="auto.commit.interval.ms" value="1000"/><entry key="session.timeout.ms" value="30000"/><entry key="key.deserializer" value="org.apache.kafka.common.serialization.StringDeserializer"/><entry key="value.deserializer" value="com.redxun.jms.ObjectDeSerializer"/></map></constructor-arg></bean><!-- 创建consumerFactory bean --><bean id="consumerFactory" class="org.springframework.kafka.core.DefaultKafkaConsumerFactory"><constructor-arg><ref bean="consumerProperties"/></constructor-arg></bean>

注意 这个分组的ID 是 group1

分组2

<bean id="consumerProperties2" class="java.util.HashMap"><constructor-arg><map><!-- 配置kafka的broke --><entry key="bootstrap.servers" value="${kafka.brokerurl}"/><!-- 配置组--><entry key="group.id" value="group2"/><entry key="enable.auto.commit" value="true"/><entry key="auto.commit.interval.ms" value="1000"/><entry key="session.timeout.ms" value="30000"/><entry key="key.deserializer" value="org.apache.kafka.common.serialization.StringDeserializer"/><entry key="value.deserializer" value="com.redxun.jms.ObjectDeSerializer"/></map></constructor-arg></bean><!-- 创建consumerFactory bean --><bean id="consumerFactory2" class="org.springframework.kafka.core.DefaultKafkaConsumerFactory"><constructor-arg><ref bean="consumerProperties2"/></constructor-arg></bean>

这里配置的分组是2 group2 。

我们使用代码测试发布消息：

IMessageProducer producer= MessageUtil.getProducer();LogEntity ent=new LogEntity();ent.setId("000000001");ent.setIp("192.168.1.1");ent.setAction("test");producer.send("logMessageQueue", ent);return "1";

我们在发布一个消息的时候，两个分组的消费者都读取到了这条消息，因此就实现了 发布订阅者模式。