赞
踩
- 1、概念:SpringBoot 整合消息服务
-
- 2、具体内容
-
- 对于异步消息组件在实际的应用之中会有两类:
-
- · JMS:代表作就是 ActiveMQ,但是其性能不高,因为其是用 java 程序实现的;
-
- · AMQP:直接利用协议实现的消息组件,其大众代表作:RabbitMQ,高性能代表作:Kafka。
-
- 2.1、SpringBoot 整合 ActiveMQ
-
- 1、 如果要想在项目之中去使用 ActiveMQ 组件,则应该为项目添加依赖支持库,修改 pom.xml 配置文件:
-
- <dependency>
- <groupId>org.springframework.boot</groupId>
- <artifactId>spring-boot-starter-activemq</artifactId>
- </dependency>
-
- 2、 修改 application.properties 配置文件进行 activemq 的配置;
-
- spring.jms.pub-sub-domain=false
- spring.activemq.user=admin
- spring.activemq.password=1234
- spring.activemq.broker-url=tcp://59.115.158.145:61616
-
- 3、 随后定义一个消息的消费者,消费者主要是进行一个监听控制,在 SpringBoot 里面可以直接利用
- 注解@JmsListener进行监听:
-
- package com.microboot.consumer;
-
- import org.springframework.jms.annotation.JmsListener;
- import org.springframework.stereotype.Service;
-
- @Service
- public class MessageConsumerService {
- @JmsListener(destination="study.msg.queue")
- public void receiveMessage(String text) { // 进行消息接收处理
- System.err.println("【*** 接收消息 ***】" + text);
- }
- }
- 4、 随后建立消息的发送者服务,一般而言如果进行消息的发送往往会准备出一个业务接口来:
-
- package com.microboot.producer;
-
- public interface IMessageProducerService {
- public void sendMessage(String msg) ;
- }
- 5、 随后建立一个配置程序类,定义 ActiveMQ 的消息发送模版处理类:
-
- package com.microboot.config;
-
- import javax.jms.Queue;
-
- import org.apache.activemq.command.ActiveMQQueue;
- import org.springframework.context.annotation.Bean;
- import org.springframework.context.annotation.Configuration;
- import org.springframework.jms.annotation.EnableJms;
-
- @Configuration
- @EnableJms
- public class ActiveMQConfig {
- @Bean
- public Queue queue() {
- return new ActiveMQQueue("study.msg.queue");
- }
- }
- 6、 创建消息发送的子类实现消息发送处理:
-
- package com.microboot.producer;
-
- import javax.annotation.Resource;
- import javax.jms.Queue;
-
- import org.springframework.jms.core.JmsMessagingTemplate;
- import org.springframework.stereotype.Service;
-
- @Service
- public class MessageProducerServiceImpl implements IMessageProducerService {
- @Resource
- private JmsMessagingTemplate jmsMessagingTemplate;
- @Resource
- private Queue queue;
- @Override
- public void sendMessage(String msg) {
- this.jmsMessagingTemplate.convertAndSend(this.queue, msg);
- }
-
- }
- 7、 编写测试类来观察消息的处理:
-
- package com.microboot.test;
-
- import javax.annotation.Resource;
-
- import org.junit.Test;
- import org.junit.runner.RunWith;
- import org.springframework.boot.test.context.SpringBootTest;
- import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
- import org.springframework.test.context.web.WebAppConfiguration;
-
- import com.microboot.StartSpringBootMain;
- import com.microboot.producer.IMessageProducerService;
-
- @SpringBootTest(classes = StartSpringBootMain.class)
- @RunWith(SpringJUnit4ClassRunner.class)
- @WebAppConfiguration
- public class TestActiveMQ {
- @Resource
- private IMessageProducerService messageProducer;
- @Test
- public void testSend() throws Exception {
- for (int x = 0; x < 10; x++) {
- this.messageProducer.sendMessage("study - " + x);
- }
- }
- }
-
- 基于 SpringBoot 配置的 JMS 的组件访问整体的处理十分简单
- 2.2、SpringBoot 整合 RabbitMQ
-
- 如果要进行 RabbitMQ 整合的时候一定要注意以下几个概念:交换空间、虚拟主机、队列信息。
-
- 本次为了方便起见将项目分为 两个:RabbitMQ-Consumer、RabbitMQ-Producer。
-
- 1、 【两个项目】将 rabbitmq 的依赖支持包拷贝到项目之中;
-
- <dependency>
- <groupId>org.springframework.boot</groupId>
- <artifactId>spring-boot-starter-amqp</artifactId>
- </dependency>
-
- 2、 【microboot-rabbitmq-producer、microboot-rabbitmq-consumer】
-
- 修改 application.properties配置文件,追加 rabbitmq 的相关配置项:
-
- spring.rabbitmq.addresses=127.0.0.1
- spring.rabbitmq.username=admin
- spring.rabbitmq.password=admin
- 3、 【microboot-rabbitmq-producer】建立一个消息的发送接口:
-
- package cn.study.microboot.producer;
-
- public interface IMessageProducerService {
- public void sendMessage(String msg) ;
- }
- 4、 【microboot-rabbitmq-producer】为了可以正常使用 RabbitMQ 进行消息处理,你还需要做一个消息
-
- 生产配置类;
-
- package cn.study.microboot.config;
-
- import org.springframework.amqp.core.Binding;
- import org.springframework.amqp.core.BindingBuilder;
- import org.springframework.amqp.core.DirectExchange;
- import org.springframework.amqp.core.Queue;
- import org.springframework.context.annotation.Bean;
- import org.springframework.context.annotation.Configuration;
-
- @Configuration
- public class ProducerConfig {
- public static final String EXCHANGE = "study.microboot.exchange"; // 交换空间名称
- public static final String ROUTINGKEY = "study.microboot.routingkey"; // 设置路由key
- public static final String QUEUE_NAME = "study.microboot.queue"; // 队列名称
- @Bean
- public Binding bindingExchangeQueue(DirectExchange exchange,Queue queue) {
- return BindingBuilder.bind(queue).to(exchange).with(ROUTINGKEY) ;
- }
- @Bean
- public DirectExchange getDirectExchange() { // 使用直连的模式
- return new DirectExchange(EXCHANGE, true, true);
- }
- @Bean
- public Queue queue() { // 要创建的队列信息
- return new Queue(QUEUE_NAME);
- }
- }
- 5、 【microboot-rabbitmq-producer】创建消息服务的实现子类:
-
- package cn.study.microboot.producer.impl;
-
- import javax.annotation.Resource;
-
- import org.springframework.amqp.rabbit.core.RabbitTemplate;
- import org.springframework.stereotype.Service;
-
- import cn.study.microboot.config.ProducerConfig;
- import cn.study.microboot.producer.IMessageProducerService;
-
- @Service
- public class MessageProducerServiceImpl implements IMessageProducerService {
- @Resource
- private RabbitTemplate rabbitTemplate;
- @Override
- public void sendMessage(String msg) {
- this.rabbitTemplate.convertAndSend(ProducerConfig.EXCHANGE,
- ProducerConfig.ROUTINGKEY, msg);
- }
-
- }
- 6、 【microboot-rabbitmq-consumer】依然需要做一个消费者的配置程序类,而这个程序类里面主要的目的
-
- 依然是设置交换空间、 路由 KEY 等信息。
-
- package cn.study.microboot.config;
-
- import org.springframework.amqp.core.Binding;
- import org.springframework.amqp.core.BindingBuilder;
- import org.springframework.amqp.core.DirectExchange;
- import org.springframework.amqp.core.Queue;
- import org.springframework.context.annotation.Bean;
- import org.springframework.context.annotation.Configuration;
-
- @Configuration
- public class ConsumerConfig {
- public static final String EXCHANGE = "study.microboot.exchange"; // 交换空间名称
- public static final String ROUTINGKEY = "study.microboot.routingkey"; // 设置路由key
- public static final String QUEUE_NAME = "study.microboot.queue"; // 队列名称
- @Bean
- public Queue queue() { // 要创建的队列信息
- return new Queue(QUEUE_NAME);
- }
- @Bean
- public DirectExchange getDirectExchange() { // 使用直连的模式
- return new DirectExchange(EXCHANGE, true, true);
- }
- @Bean
- public Binding bindingExchangeQueue(DirectExchange exchange,Queue queue) {
- return BindingBuilder.bind(queue).to(exchange).with(ROUTINGKEY) ;
- }
- }
- 7、 【microboot-rabbitmq-consumer】实现监听处理类:
-
- package cn.study.microboot.consumer;
-
- import org.springframework.amqp.rabbit.annotation.RabbitListener;
- import org.springframework.stereotype.Service;
-
- @Service
- public class MessageConsumerService {
- @RabbitListener(queues="study.microboot.queue")
- public void receiveMessage(String text) { // 进行消息接收处理
- System.err.println("【*** 接收消息 ***】" + text);
- }
- }
-
- 9、 【microboot-rabbitmq-consumer】编写消息接收测试类,这里面不需要编写代码,只需要做一个休眠
-
- 即可:
-
- package cn.study.microboot.test;
-
- import org.junit.Test;
- import org.junit.runner.RunWith;
- import org.springframework.boot.test.context.SpringBootTest;
- import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
- import org.springframework.test.context.web.WebAppConfiguration;
-
- import cn.study.microboot.StartSpringBootMain;
-
- @SpringBootTest(classes = StartSpringBootMain.class)
- @RunWith(SpringJUnit4ClassRunner.class)
- @WebAppConfiguration
- public class AppTest {
- @Test
- public void testStart() throws Exception {
- Thread.sleep(Long.MAX_VALUE);
- }
- }
-
- 整体进行项目开发之中整合的处理步骤还是简单,但是千万要注意,由于是第一次整合处理,所以将生产者与
-
- 消费者的配置 类分开了,实际上这两个类的作用是完全一样的。
- 2.3、SpringBoot 整合 Kafka
-
- Kafka 是现在最好的开源消息组件,其仿照 AMQP 协议操作,而且处理的性能也是最高的。本次使用已经配置
-
- 好的 Kafka 服 务器,而且这台服务器上使用了 kerberos 认证,所以应该首先准备好一个 jass 配置文件:
-
- 1、 定义“kafka_client_jaas.conf”配置文件:
-
- KafkaClient {
- org.apache.kafka.common.security.plain.PlainLoginModule required
- username="bob"
- password="bob-pwd";
- };
- 2、 为了方便进行项目的观察, 本次依然准备出了两个项目:生产者( microboot-kafka-producer )、
-
- 消 费 者 (microboot-kafka-consumer),随后为这两个项目添加 kafka 配置支持:
-
- <dependency>
- <groupId>org.springframework.kafka</groupId>
- <artifactId>spring-kafka</artifactId>
- </dependency>
- 3、 【micorboot-kafka-consumer】修改 application.yml 配置文件,进行 kafka 配置项编写:
-
- server:
- port: 80
- spring:
- messages:
- basename: i18n/Messages,i18n/Pages
- kafka:
- bootstrap-servers:
- - kafka-single:9095
- template:
- default-topic: mldn-microboot
- consumer:
- key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
- value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
- group-id: group-1
- properties:
- sasl.mechanism: PLAIN
- security.protocol: SASL_PLAINTEXT
-
- 4、 【micorboot-kafka-consumer】建立一个 Kafka 的消息的消费程序类:
-
- package cn.study.microboot.consumer;
-
- import org.apache.kafka.clients.consumer.ConsumerRecord;
- import org.springframework.kafka.annotation.KafkaListener;
- import org.springframework.stereotype.Service;
-
- @Service
- public class MessageConsumerService {
- @KafkaListener(topics = {"study-microboot"})
- public void receiveMessage(ConsumerRecord<String, String> record) { // 进行消息接收处理
- System.err.println("【*** 接收消息 ***】key = " + record.key() + "、value = "
- + record.value());
- }
- }
- 5、 【micorboot-kafka-consumer】随后还需要修改 SpringBoot 的启动程序类,追加 kerberos 配置:
-
- package cn.study.microboot;
-
- import org.springframework.boot.SpringApplication;
- import org.springframework.boot.autoconfigure.SpringBootApplication;
-
- @SpringBootApplication // 启动SpringBoot程序,而后自带子包扫描
- public class StartSpringBootMain {
- static {
- System.setProperty("java.security.auth.login.config",
- "d:/kafka_client_jaas.conf"); // 表示系统环境属性
- }
- public static void main(String[] args) throws Exception {
- SpringApplication.run(StartSpringBootMain.class, args);
- }
- }
- 6、 【microboot-kafka-producer】修改 application.yml 配置文件:
-
- server:
- port: 80
- spring:
- messages:
- basename: i18n/Messages,i18n/Pages
- kafka:
- bootstrap-servers:
- - kafka-single:9095
- template:
- default-topic: mldn-microboot
- producer:
- key-serializer: org.apache.kafka.common.serialization.StringSerializer
- value-serializer: org.apache.kafka.common.serialization.StringSerializer
- properties:
- sasl.mechanism: PLAIN
- security.protocol: SASL_PLAINTEXT
- 7、 【microboot-kafka-producer】定义消息发送的服务接口:
-
- package cn.study.microboot.producer;
-
- public interface IMessageProducerService {
- public void sendMessage(String msg) ;
- }
-
- package cn.study.microboot.service.impl;
-
- import javax.annotation.Resource;
-
- import org.springframework.kafka.core.KafkaTemplate;
- import org.springframework.stereotype.Service;
-
- import cn.study.microboot.service.IMessageProducerService;
-
- @Service
- public class MessageProducerServiceImpl implements IMessageProducerService {
- @Resource
- private KafkaTemplate<String, String> kafkaTemplate;
- @Override
- public void send(String msg) {
- this.kafkaTemplate.sendDefault("study-key", msg);
- }
-
- }
- 8、 【microboot-kafka-producer】修改程序启动类:
-
- package cn.mldn.microboot;
-
- import org.springframework.boot.SpringApplication;
- import org.springframework.boot.autoconfigure.SpringBootApplication;
-
- @SpringBootApplication // 启动SpringBoot程序,而后自带子包扫描
- public class StartSpringBootMain {
- static {
- System.setProperty("java.security.auth.login.config",
- "d:/kafka_client_jaas.conf"); // 表示系统环境属性
- }
- public static void main(String[] args) throws Exception {
- SpringApplication.run(StartSpringBootMain.class, args);
- }
- }
- 9、 【microboot-kafka-producer】编写消息发送的程序类:
-
- package cn.study.microboot;
-
- import javax.annotation.Resource;
-
- import org.junit.Test;
- import org.junit.runner.RunWith;
- import org.springframework.boot.test.context.SpringBootTest;
- import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
- import org.springframework.test.context.web.WebAppConfiguration;
-
- import cn.study.microboot.service.IMessageProducerService;
- @SpringBootTest(classes = StartSpringBootMain.class)
- @RunWith(SpringJUnit4ClassRunner.class)
- @WebAppConfiguration
- public class TestMessageService {
- @Resource
- private IMessageProducerService messageService;
- @Test
- public void testStart() throws Exception {
- for (int x = 0; x < 100; x++) {
- this.messageService.send("study - " + x);
- }
- }
- }
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。