spring boot + kafka 使用详细步骤
2017-08-17 18:10
537 查看
一,Kafka的安装以及配置
1.下载文件wget http://mirror.bit.edu.cn/apache/kafka/0.11.0.0/kafka_2.11-0.11.0.0.tgz
2.安装
tar xzvf kafka_2.11-0.11.0.0.tgz -C /usr/local/
3.配置(服务器在阿里云ECS上)
vi %kafka_home%/server.properties ############################# Server Basics ############################# broker.id=0 port=9092 host.name=阿里云内网ip advertised.host.name=阿里云外网ip delete.topic.enable=true ....其他的配置不变
二,Spring boot操作Kafka
1.Spring boot pom依赖<parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>1.5.6.RELEASE</version> <relativePath/> <!-- lookup parent from repository --> </parent>
2.kafka依赖
<!--kafka支持--> <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency>
3.kafka配置
#kafka相关配置 spring.kafka.bootstrap-servers=阿里云外网ip:9092 #设置一个默认组 spring.kafka.consumer.group-id=defaultGroup #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
4.kafka发送代码
@Component public class KafkaSender { @Autowired private KafkaTemplate kafkaTemplate; /** * 发送消息到kafka,主题为test */ public void sendTest(){ kafkaTemplate.send("test","hello,kafka " + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS"))); } }
5.Kafka消费代码
public class KafkaConsumer { /** * 监听test主题,有消息就读取 * @param message */ @KafkaListener(topics = {"test"}) public void consumer(String message){ log.info("test topic message : {}", message); } }
6,启动测试代码
@SpringBootApplication @EnableScheduling public class ServerApplication { @Autowired private KafkaSender kafkaSender; public static void main(String[] args) { SpringApplication.run(ServerApplication.class, args); } //然后每隔1分钟执行一次 @Scheduled(fixedRate = 1000 * 60) public void testKafka() throws Exception { kafkaSender.sendTest(); } }
三,以上就是完整的spring boot整合Kafka消息队列的步骤.
相关文章推荐
- spring boot + kafka 使用详细步骤
- [置顶] 使用SpringBoot开启微服务之旅(详细步骤)
- Spring-Boot使用Spring-Kafka生产者
- Spring Boot 中使用 kafka
- springboot框架中使用java操作kafka获取数据
- kafka分布式消息队列使用(springboot和springmvc)
- 详细步骤!!!idea+springboot+mybatis+jsp+bootstrap实现从mysql查询出数据并显示(源码)
- SpringBoot之瘦身部署的详细步骤
- 使用IDEA创建SpringBoot项目的方法步骤
- spring boot 集成kafka (多线程,消费者使用kafka的原生api实现,因为@KakfkaListener修改groupId无效)
- kafka分布式消息队列使用(springboot和springmvc)
- 超级详细的新手在Idea中使用maven配置Spring+springmvc+mybatis(SSM)框架步骤
- spring boot 使用logback日志系统的详细说明
- Springboot搭建dubbo(详细,zookeeper使用windows版)
- Spring boot 下使用Kafka实例
- eclipse+springboot+maven访问jsp页面,详细步骤图解!!!(源码)
- SpringBoot项目中使用redis缓存的方法步骤
- Springboot项目使用aop切面保存详细日志到ELK日志平台
- 【spring boot】11.spring-data-jpa的详细介绍和复杂使用
- Spring-Boot 集成Solr客户端的详细步骤