网络技术

找论坛
mb46348043
Lv2 太平洋舰队下士
太平洋舰队下士 贡献175,距离下一级还需25贡献
楼主
2018-02-09 09:50 0 0 只看楼主
电梯直达 
  Spring Boot 入门之消息中间件的使用
  在消息中间件中有 2 个重要的概念:消息代理和目的地。当消息发送者发送消息后,消息就被消息代理接管,消息代理保证消息传递到指定目的地。
  我们常用的消息代理有 JMS 和 AMQP 规范。对应地,它们常见的实现分别是 ActiveMQ 和 RabbitMQ。
  二、整合 ActiveMQ
  2.1 添加依赖
  ?
  org.springframework.boot
  spring-boot-starter-activemq
  org.apache.activemq
  activemq-pool
  2.2 添加配置
  ?
  8# activemq 配置
  spring.activemq.broker-url=tcp://192.168.2.12:61616
  spring.activemq.user=admin
  spring.activemq.password=admin
  spring.activemq.pool.enabled=false
  spring.activemq.pool.max-connections=50
  # 使用发布/订阅模式时,下边配置需要设置成 true
  spring.jms.pub-sub-domain=false
  此处 spring.activemq.pool.enabled=false,表示关闭连接池。
  2.3 编码
  配置类:
  ?
  16@Configuration
  public class JmsConfirguration {
  public static final String QUEUE_NAME = "activemq_queue";
  public static final String TOPIC_NAME = "activemq_topic";
  @Bean
  public Queue queue() {
  return new ActiveMQQueue(QUEUE_NAME);
  }
  @Bean
  public Topic topic() {
  return new ActiveMQTopic(TOPIC_NAME);
  }
  }
  负责创建队列和主题。
  消息生产者:
  ?
  19@Component
  public class JmsSender {
  @Autowired
  private Queue queue;
  @Autowired
  private Topic topic;
  @Autowired
  private JmsMessagingTemplate jmsTemplate;
  public void sendByQueue(String message) {
  this.jmsTemplate.convertAndSend(queue, message);
  }
  public void sendByTopic(String message) {
  this.jmsTemplate.convertAndSend(topic, message);
  }
  }
  消息消费者:
  ?
  13@Component
  public class JmsReceiver {
  @JmsListener(destination = JmsConfirguration.QUEUE_NAME)
  public void receiveByQueue(String message) {
  System.out.println("接收队列消息:" + message);
  }
  @JmsListener(destination = JmsConfirguration.TOPIC_NAME)
  public void receiveByTopic(String message) {
  System.out.println("接收主题消息:" + message);
  }
  }
  消息消费者使用 @JmsListener 注解监听消息。
  2.4 测试
  ?
  19@RunWith(SpringRunner.class)
  @SpringBootTest
  public class JmsTest {
  @Autowired
  private JmsSender sender;
  @Test
  public void testSendByQueue() {
  for (int i = 1; i < 6; i++) {
  this.sender.sendByQueue("hello activemq queue " + i);
  }
  }
  @Test
  public void testSendByTopic() {
  for (int i = 1; i < 6; i++) {
  this.sender.sendByTopic("hello activemq topic " + i);
  }
  }
  }
  打印结果:
  接收队列消息:hello activemq queue 1
  接收队列消息:hello activemq queue 2
  接收队列消息:hello activemq queue 3
  接收队列消息:hello activemq queue 4
  接收队列消息:hello activemq queue 5
  测试发布/订阅模式时,设置 spring.jms.pub-sub-domain=true
  接收主题消息:hello activemq topic 1
  接收主题消息:hello activemq topic 2
  接收主题消息:hello activemq topic 3
  接收主题消息:hello activemq topic 4
  接收主题消息:hello activemq topic 5
  三、整合 RabbitMQ
  3.1 添加依赖
  ?
  org.springframework.boot
  spring-boot-starter-amqp
  3.2 添加配置
  ?
  5spring.rabbitmq.host=192.168.2.30
  spring.rabbitmq.port=5672
  spring.rabbitmq.username=light
  spring.rabbitmq.password=light
  spring.rabbitmq.virtual-host=/test
  3.3 编码
  配置类:
  ?
  102@Configuration
  public class AmqpConfirguration {
  //=============简单、工作队列模式===============
  public static final String SIMPLE_QUEUE = "simple_queue";
  @Bean
  public Queue queue() {
  return new Queue(SIMPLE_QUEUE, true);
  }
  //===============发布/订阅模式============
  public static final String PS_QUEUE_1 = "ps_queue_1";
  public static final String PS_QUEUE_2 = "ps_queue_2";
  public static final String FANOUT_EXCHANGE = "fanout_exchange";
  @Bean
  public Queue psQueue1() {
  return new Queue(PS_QUEUE_1, true);
  }
  @Bean
  public Queue psQueue2() {
  return new Queue(PS_QUEUE_2, true);
  }
  @Bean
  public FanoutExchange fanoutExchange() {
  return new FanoutExchange(FANOUT_EXCHANGE);
  }
  @Bean
  public Binding fanoutBinding1() {
  return BindingBuilder.bind(psQueue1()).to(fanoutExchange());
  }
  @Bean
  public Binding fanoutBinding2() {
  return BindingBuilder.bind(psQueue2()).to(fanoutExchange());
  }
  //===============路由模式============
  public static final String ROUTING_QUEUE_1 = "routing_queue_1";
  public static final String ROUTING_QUEUE_2 = "routing_queue_2";
  public static final String DIRECT_EXCHANGE = "direct_exchange";
  @Bean
  public Queue routingQueue1() {
  return new Queue(ROUTING_QUEUE_1, true);
  }
  @Bean
  public Queue routingQueue2() {
  return new Queue(ROUTING_QUEUE_2, true);
  }
  @Bean
  public DirectExchange directExchange() {
  return new DirectExchange(DIRECT_EXCHANGE);
  }
  @Bean
  public Binding directBinding1() {
  return BindingBuilder.bind(routingQueue1()).to(directExchange()).with("user");
  }
  @Bean
  public Binding directBinding2() {
  return BindingBuilder.bind(routingQueue2()).to(directExchange()).with("order");
  }
  //===============主题模式============
  public static final String TOPIC_QUEUE_1 = "topic_queue_1";
  public static final String TOPIC_QUEUE_2 = "topic_queue_2";
  public static final String TOPIC_EXCHANGE = "topic_exchange";
  @Bean
  public Queue topicQueue1() {
  return new Queue(TOPIC_QUEUE_1, true);
  }
  @Bean
  public Queue topicQueue2() {
  return new Queue(TOPIC_QUEUE_2, true);
  }
  @Bean
  public TopicExchange topicExchange() {
  return new TopicExchange(TOPIC_EXCHANGE);
  }
  @Bean
  public Binding topicBinding1() {
  return BindingBuilder.bind(topicQueue1()).to(topicExchange()).with("user.add");
  }
  @Bean
  public Binding topicBinding2() {
  return BindingBuilder.bind(topicQueue2()).to(topicExchange()).with("user.#");
  }
  }
  RabbitMQ 有多种工作模式,因此配置比较多。想了解谭八爷系统相关内容的读者可以查看《RabbitMQ 工作模式介绍》或者自行百度相关资料。
  消息生产者:
  ?
  38@Component
  public class AmqpSender {
  @Autowired
  private AmqpTemplate amqpTemplate;
  /**
  * 简单模式发送
  *
  * @param message
  */
  public void simpleSend(String message) {
  this.amqpTemplate.convertAndSend(AmqpConfirguration.SIMPLE_QUEUE, message);
  }
  /**
  * 发布/订阅模式发送
  *
  * @param message
  */
  public void psSend(String message) {
  this.amqpTemplate.convertAndSend(AmqpConfirguration.FANOUT_EXCHANGE, "", message);
  }
  /**
  * 路由模式发送
  *
  * @param message
  */
  public void routingSend(String routingKey, String message) {
  this.amqpTemplate.convertAndSend(AmqpConfirguration.DIRECT_EXCHANGE, routingKey, message);
  }
  /**
  * 主题模式发送
  *
  * @param routingKey
  * @param message
  */
  public void topicSend(String routingKey, String message) {
  this.amqpTemplate.convertAndSend(AmqpConfirguration.TOPIC_EXCHANGE, routingKey, message);
  }
  }
  消息消费者:
  ?
  52@Component
  public class AmqpReceiver {
  /**
  * 简单模式接收
  *
  * @param message
  */
  @RabbitListener(queues = AmqpConfirguration.SIMPLE_QUEUE)
  public void simpleReceive(String message) {
  System.out.println("接收消息:" + message);
  }
  /**
  * 发布/订阅模式接收
  *
  * @param message
  */
  @RabbitListener(queues = AmqpConfirguration.PS_QUEUE_1)
  public void psReceive1(String message) {
  System.out.println(AmqpConfirguration.PS_QUEUE_1 + "接收消息:" + message);
  }
  @RabbitListener(queues = AmqpConfirguration.PS_QUEUE_2)
  public void psReceive2(String message) {
  System.out.println(AmqpConfirguration.PS_QUEUE_2 + "接收消息:" + message);
  }
  /**
  * 路由模式接收
  *
  * @param message
  */
  @RabbitListener(queues = AmqpConfirguration.ROUTING_QUEUE_1)
  public void routingReceive1(String message) {
  System.out.println(AmqpConfirguration.ROUTING_QUEUE_1 + "接收消息:" + message);
  }
  @RabbitListener(queues = AmqpConfirguration.ROUTING_QUEUE_2)
  public void routingReceive2(String message) {
  System.out.println(AmqpConfirguration.ROUTING_QUEUE_2 + "接收消息:" + message);
  }
  /**
  * 主题模式接收
  *
  * @param message
  */
  @RabbitListener(queues = AmqpConfirguration.TOPIC_QUEUE_1)
  public void topicReceive1(String message) {
  System.out.println(AmqpConfirguration.TOPIC_QUEUE_1 + "接收消息:" + message);
  }
  @RabbitListener(queues = AmqpConfirguration.TOPIC_QUEUE_2)
  public void topicReceive2(String message) {
  System.out.println(AmqpConfirguration.TOPIC_QUEUE_2 + "接收消息:" + message);
  }
  }
  消息消费者使用 @RabbitListener 注解监听消息。
  3.4 测试
  ?
  32@RunWith(SpringRunner.class)
  @SpringBootTest
  public class AmqpTest {
  @Autowired
  private AmqpSender sender;
  @Test
  public void testSimpleSend() {
  for (int i = 1; i < 6; i++) {
  this.sender.simpleSend("test simpleSend " + i);
  }
  }
  @Test
  public void testPsSend() {
  for (int i = 1; i < 6; i++) {
  this.sender.psSend("test psSend " + i);
  }
  }
  @Test
  public void testRoutingSend() {
  for (int i = 1; i < 6; i++) {
  this.sender.routingSend("order", "test routingSend " + i);
  }
  }
  @Test
  public void testTopicSend() {
  for (int i = 1; i < 6; i++) {
  this.sender.topicSend("user.add", "test topicSend " + i);
  }
  }
  }
  测试结果略过。。。
  踩坑提醒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 选项中手动添加对应的队列。

您需要登录后才可以发帖 登录 | 立即注册

其他登录方式:

常用表情
太平洋电脑网论坛帖子仅代表作者本人意见,不代表网站立场。请勿轻信特价、汇款、中奖等信息,
请勿轻易透露个人资料,因此产生的一切后果,PConline不承担任何责任
回复 发新帖 找论坛 反馈 回顶部