每个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){ //收到通道的消息之后执行秒杀操作 } }