闽公网安备 35020302035485号

每个Consumer属于一个特定的Consumer Group(可为每个Consumer指定group name,若不指定group name则属于默认的group)。
wget http://mirror.bit.edu.cn/apache/kafka/1.1.0/kafka_2.11-1.1.0.tgz将包下载到执行目录并解压:
cd /usr/local/ tar -xzvf kafka_2.11-0.10.0.1.tgz修改kafka配置文件:
cd kafka_2.11-0.10.0.1/config/ #编辑配置文件 vi server.properties broker.id=0 #端口号、记得开启端口,云服务器要开放安全组 port=9092 #服务器IP地址,修改为自己的服务器IP host.name=127.0.0.1 #zookeeper地址和端口, Kafka支持内置的Zookeeper和引用外部的Zookeeper zookeeper.connect=localhost:2181分别启动 kafka 和 zookeeper:
./zookeeper-server-start.sh /usr/local/kafka_2.11-0.10.0.1/config/zookeeper.properties & ./kafka-server-start.sh /usr/local/kafka_2.11-0.10.0.1/config/server.properties &SpringBoot集成
<!--堆代码 duidaima.com-->
<!--kafka支持-->
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<version>1.3.5.RELEASE</version><!--$NO-MVN-MAN-VER$-->
</dependency>
application.properties配置:#kafka相关配置 spring.kafka.bootstrap-servers=192.168.1.180:9092 #设置一个默认组 spring.kafka.consumer.group-id=0 #key-value序列化反序列化 spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer #每次批量发送消息的数量 spring.kafka.producer.batch-size=65536 spring.kafka.producer.buffer-memory=524288生产者KafkaSender:
/**
* 堆代码 duidaima.com
* 生产者
* @author 科帮网 By https://blog.52itstyle.vip
*/
@Component
public class KafkaSender {
@Autowired
private KafkaTemplate<String,String> kafkaTemplate;
/**
* 发送消息到kafka
*/
public void sendChannelMess(String channel, String message){
kafkaTemplate.send(channel,message);
}
}
消费者:
/**
* 消费者 spring-kafka 2.0 + 依赖JDK8
* @author 科帮网 By https://blog.52itstyle.vip
*/
@Component
public class KafkaConsumer {
/**
* 监听seckill主题,有消息就读取
* @param message
*/
@KafkaListener(topics = {"seckill"})
public void receiveMessage(String message){
//收到通道的消息之后执行秒杀操作
}
}