一、前言

在消息中间件中有 2 个重要的概念:消息代理和目的地。当消息发送者发送消息后,消息就被消息代理接管,消息代理保证消息传递到指定目的地。

我们常用的消息代理有 JMS 和 AMQP 规范。对应地,它们常见的实现分别是 ActiveMQ 和 RabbitMQ。

上篇文章《Spring Boot 入门之缓存和 NoSQL 篇(四)》

二、整合 ActiveMQ

# 2.1 添加依赖

  1. <dependency>
  2. <groupId>org.springframework.boot</groupId>
  3. <artifactId>spring-boot-starter-activemq</artifactId>
  4. </dependency>
  5. <!-- 如果需要配置连接池,添加如下依赖 -->
  6. <dependency>
  7. <groupId>org.apache.activemq</groupId>
  8. <artifactId>activemq-pool</artifactId>
  9. </dependency>

# 2.2 添加配置

  1. # activemq 配置
  2. spring.activemq.broker-url=tcp://192.168.2.61:61616
  3. spring.activemq.user=admin
  4. spring.activemq.password=admin
  5. spring.activemq.pool.enabled=false
  6. spring.activemq.pool.max-connections=50
  7. # 使用发布/订阅模式时,下边配置需要设置成 true
  8. spring.jms.pub-sub-domain=false

此处 spring.activemq.pool.enabled=false,表示关闭连接池。

# 2.3 编码

配置类:

  1. @Configuration
  2. public class JmsConfirguration {
  3. public static final String QUEUE_NAME = "activemq_queue";
  4. public static final String TOPIC_NAME = "activemq_topic";
  5. @Bean
  6. public Queue queue() {
  7. return new ActiveMQQueue(QUEUE_NAME);
  8. }
  9. @Bean
  10. public Topic topic() {
  11. return new ActiveMQTopic(TOPIC_NAME);
  12. }
  13. }

负责创建队列和主题。

消息生产者:

  1. @Component
  2. public class JmsSender {
  3. @Autowired
  4. private Queue queue;
  5. @Autowired
  6. private Topic topic;
  7. @Autowired
  8. private JmsMessagingTemplate jmsTemplate;
  9. public void sendByQueue(String message) {
  10. this.jmsTemplate.convertAndSend(queue, message);
  11. }
  12. public void sendByTopic(String message) {
  13. this.jmsTemplate.convertAndSend(topic, message);
  14. }
  15. }

消息消费者:

  1. @Component
  2. public class JmsReceiver {
  3. @JmsListener(destination = JmsConfirguration.QUEUE_NAME)
  4. public void receiveByQueue(String message) {
  5. System.out.println("接收队列消息:" + message);
  6. }
  7. @JmsListener(destination = JmsConfirguration.TOPIC_NAME)
  8. public void receiveByTopic(String message) {
  9. System.out.println("接收主题消息:" + message);
  10. }
  11. }

消息消费者使用 @JmsListener 注解监听消息。

# 2.4 测试

  1. @RunWith(SpringRunner.class)
  2. @SpringBootTest
  3. public class JmsTest {
  4. @Autowired
  5. private JmsSender sender;
  6. @Test
  7. public void testSendByQueue() {
  8. for (int i = 1; i < 6; i++) {
  9. this.sender.sendByQueue("hello activemq queue " + i);
  10. }
  11. }
  12. @Test
  13. public void testSendByTopic() {
  14. for (int i = 1; i < 6; i++) {
  15. this.sender.sendByTopic("hello activemq topic " + i);
  16. }
  17. }
  18. }

打印结果:

  1. 接收队列消息:hello activemq queue 1
  2. 接收队列消息:hello activemq queue 2
  3. 接收队列消息:hello activemq queue 3
  4. 接收队列消息:hello activemq queue 4
  5. 接收队列消息:hello activemq queue 5

测试发布/订阅模式时,设置 spring.jms.pub-sub-domain=true

  1. 接收主题消息:hello activemq topic 1
  2. 接收主题消息:hello activemq topic 2
  3. 接收主题消息:hello activemq topic 3
  4. 接收主题消息:hello activemq topic 4
  5. 接收主题消息:hello activemq topic 5

三、整合 RabbitMQ

# 3.1 添加依赖

  1. <dependency>
  2. <groupId>org.springframework.boot</groupId>
  3. <artifactId>spring-boot-starter-amqp</artifactId>
  4. </dependency>

# 3.2 添加配置

  1. spring.rabbitmq.host=192.168.2.71
  2. spring.rabbitmq.port=5672
  3. spring.rabbitmq.username=light
  4. spring.rabbitmq.password=light
  5. spring.rabbitmq.virtual-host=/test

# 3.3 编码

配置类:

  1. @Configuration
  2. public class AmqpConfirguration {
  3. //=============简单、工作队列模式===============
  4. public static final String SIMPLE_QUEUE = "simple_queue";
  5. @Bean
  6. public Queue queue() {
  7. return new Queue(SIMPLE_QUEUE, true);
  8. }
  9. //===============发布/订阅模式============
  10. public static final String PS_QUEUE_1 = "ps_queue_1";
  11. public static final String PS_QUEUE_2 = "ps_queue_2";
  12. public static final String FANOUT_EXCHANGE = "fanout_exchange";
  13. @Bean
  14. public Queue psQueue1() {
  15. return new Queue(PS_QUEUE_1, true);
  16. }
  17. @Bean
  18. public Queue psQueue2() {
  19. return new Queue(PS_QUEUE_2, true);
  20. }
  21. @Bean
  22. public FanoutExchange fanoutExchange() {
  23. return new FanoutExchange(FANOUT_EXCHANGE);
  24. }
  25. @Bean
  26. public Binding fanoutBinding1() {
  27. return BindingBuilder.bind(psQueue1()).to(fanoutExchange());
  28. }
  29. @Bean
  30. public Binding fanoutBinding2() {
  31. return BindingBuilder.bind(psQueue2()).to(fanoutExchange());
  32. }
  33. //===============路由模式============
  34. public static final String ROUTING_QUEUE_1 = "routing_queue_1";
  35. public static final String ROUTING_QUEUE_2 = "routing_queue_2";
  36. public static final String DIRECT_EXCHANGE = "direct_exchange";
  37. @Bean
  38. public Queue routingQueue1() {
  39. return new Queue(ROUTING_QUEUE_1, true);
  40. }
  41. @Bean
  42. public Queue routingQueue2() {
  43. return new Queue(ROUTING_QUEUE_2, true);
  44. }
  45. @Bean
  46. public DirectExchange directExchange() {
  47. return new DirectExchange(DIRECT_EXCHANGE);
  48. }
  49. @Bean
  50. public Binding directBinding1() {
  51. return BindingBuilder.bind(routingQueue1()).to(directExchange()).with("user");
  52. }
  53. @Bean
  54. public Binding directBinding2() {
  55. return BindingBuilder.bind(routingQueue2()).to(directExchange()).with("order");
  56. }
  57. //===============主题模式============
  58. public static final String TOPIC_QUEUE_1 = "topic_queue_1";
  59. public static final String TOPIC_QUEUE_2 = "topic_queue_2";
  60. public static final String TOPIC_EXCHANGE = "topic_exchange";
  61. @Bean
  62. public Queue topicQueue1() {
  63. return new Queue(TOPIC_QUEUE_1, true);
  64. }
  65. @Bean
  66. public Queue topicQueue2() {
  67. return new Queue(TOPIC_QUEUE_2, true);
  68. }
  69. @Bean
  70. public TopicExchange topicExchange() {
  71. return new TopicExchange(TOPIC_EXCHANGE);
  72. }
  73. @Bean
  74. public Binding topicBinding1() {
  75. return BindingBuilder.bind(topicQueue1()).to(topicExchange()).with("user.add");
  76. }
  77. @Bean
  78. public Binding topicBinding2() {
  79. return BindingBuilder.bind(topicQueue2()).to(topicExchange()).with("user.#");
  80. }
  81. }

RabbitMQ 有多种工作模式,因此配置比较多。想了解相关内容的读者可以查看本站的《RabbitMQ 工作模式介绍》或者自行百度相关资料。

消息生产者:

  1. @Component
  2. public class AmqpSender {
  3. @Autowired
  4. private AmqpTemplate amqpTemplate;
  5. /**
  6. * 简单模式发送
  7. *
  8. * @param message
  9. */
  10. public void simpleSend(String message) {
  11. this.amqpTemplate.convertAndSend(AmqpConfirguration.SIMPLE_QUEUE, message);
  12. }
  13. /**
  14. * 发布/订阅模式发送
  15. *
  16. * @param message
  17. */
  18. public void psSend(String message) {
  19. this.amqpTemplate.convertAndSend(AmqpConfirguration.FANOUT_EXCHANGE, "", message);
  20. }
  21. /**
  22. * 路由模式发送
  23. *
  24. * @param message
  25. */
  26. public void routingSend(String routingKey, String message) {
  27. this.amqpTemplate.convertAndSend(AmqpConfirguration.DIRECT_EXCHANGE, routingKey, message);
  28. }
  29. /**
  30. * 主题模式发送
  31. *
  32. * @param routingKey
  33. * @param message
  34. */
  35. public void topicSend(String routingKey, String message) {
  36. this.amqpTemplate.convertAndSend(AmqpConfirguration.TOPIC_EXCHANGE, routingKey, message);
  37. }
  38. }

消息消费者:

  1. @Component
  2. public class AmqpReceiver {
  3. /**
  4. * 简单模式接收
  5. *
  6. * @param message
  7. */
  8. @RabbitListener(queues = AmqpConfirguration.SIMPLE_QUEUE)
  9. public void simpleReceive(String message) {
  10. System.out.println("接收消息:" + message);
  11. }
  12. /**
  13. * 发布/订阅模式接收
  14. *
  15. * @param message
  16. */
  17. @RabbitListener(queues = AmqpConfirguration.PS_QUEUE_1)
  18. public void psReceive1(String message) {
  19. System.out.println(AmqpConfirguration.PS_QUEUE_1 + "接收消息:" + message);
  20. }
  21. @RabbitListener(queues = AmqpConfirguration.PS_QUEUE_2)
  22. public void psReceive2(String message) {
  23. System.out.println(AmqpConfirguration.PS_QUEUE_2 + "接收消息:" + message);
  24. }
  25. /**
  26. * 路由模式接收
  27. *
  28. * @param message
  29. */
  30. @RabbitListener(queues = AmqpConfirguration.ROUTING_QUEUE_1)
  31. public void routingReceive1(String message) {
  32. System.out.println(AmqpConfirguration.ROUTING_QUEUE_1 + "接收消息:" + message);
  33. }
  34. @RabbitListener(queues = AmqpConfirguration.ROUTING_QUEUE_2)
  35. public void routingReceive2(String message) {
  36. System.out.println(AmqpConfirguration.ROUTING_QUEUE_2 + "接收消息:" + message);
  37. }
  38. /**
  39. * 主题模式接收
  40. *
  41. * @param message
  42. */
  43. @RabbitListener(queues = AmqpConfirguration.TOPIC_QUEUE_1)
  44. public void topicReceive1(String message) {
  45. System.out.println(AmqpConfirguration.TOPIC_QUEUE_1 + "接收消息:" + message);
  46. }
  47. @RabbitListener(queues = AmqpConfirguration.TOPIC_QUEUE_2)
  48. public void topicReceive2(String message) {
  49. System.out.println(AmqpConfirguration.TOPIC_QUEUE_2 + "接收消息:" + message);
  50. }
  51. }

消息消费者使用 @RabbitListener 注解监听消息。

# 3.4 测试

  1. @RunWith(SpringRunner.class)
  2. @SpringBootTest
  3. public class AmqpTest {
  4. @Autowired
  5. private AmqpSender sender;
  6. @Test
  7. public void testSimpleSend() {
  8. for (int i = 1; i < 6; i++) {
  9. this.sender.simpleSend("test simpleSend " + i);
  10. }
  11. }
  12. @Test
  13. public void testPsSend() {
  14. for (int i = 1; i < 6; i++) {
  15. this.sender.psSend("test psSend " + i);
  16. }
  17. }
  18. @Test
  19. public void testRoutingSend() {
  20. for (int i = 1; i < 6; i++) {
  21. this.sender.routingSend("order", "test routingSend " + i);
  22. }
  23. }
  24. @Test
  25. public void testTopicSend() {
  26. for (int i = 1; i < 6; i++) {
  27. this.sender.topicSend("user.add", "test topicSend " + i);
  28. }
  29. }
  30. }

测试结果略过。。。

踩坑提醒1:ACCESS_REFUSED - Login was refused using authentication mechanism PLAIN

解决方案:

  1. 请确保用户名和密码是否正确,需要注意的时用户名和密码的值是否包含空格或制表符。

  2. 如果测试账户使用的是 guest,需要修改 rabbitmq.conf 文件。在该文件中添加 “loopback_users = none” 配置。

踩坑提醒2:Cannot prepare queue for listener. Either the queue doesn't exist or the broker will not allow us to use it

解决方案:

我们可以登陆 RabbitMQ 的管理界面,在 Queue 选项中手动添加对应的队列。

四、源码下载

五、参考资料