From d85a656874282c6aafda897e2249c2e078e9363e Mon Sep 17 00:00:00 2001 From: wangkang Date: Wed, 14 Jun 2017 15:01:23 +0800 Subject: [PATCH] =?UTF-8?q?=E6=9B=B4=E6=96=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- 9.md | 174 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 174 insertions(+) create mode 100644 9.md diff --git a/9.md b/9.md new file mode 100644 index 0000000..79a6ce0 --- /dev/null +++ b/9.md @@ -0,0 +1,174 @@ +# spring boot / cloud (九) 使用rabbitmq消息中间件 + +##前言 + + +###rabbitmq介绍: + + +>RabbitMQ是一个在AMQP基础上完整的，可复用的企业消息系统。它可以用于大型软件系统各个模块之间的高效通信，支持高并发，支持可扩展。 + + +###amqp介绍: + + +>即Advanced Message Queuing Protocol,一个提供统一消息服务的应用层标准高级消息队列协议,是应用层协议的一个开放标准,为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同的开发语言等条件的限制。Erlang中的实现有 RabbitMQ等。 + + +##思路 + + +基于spring boot的特性连接rabbitmq,并作出如下样例: + + +- 配置 + + +- 发布方样例 + + +- 消费方样例 + + +##实现 + + +###1.配置 + +引入maven依赖 + +``` xml + + org.springframework.cloud + spring-cloud-starter-bus-amqp + + + + org.springframework.boot + spring-boot-starter-amqp + +``` + +编写config配置类(默认情况下是不用做任何配置的,这里有配置是因为,它默认是用的二进制做的消息传输,这里的配置是改为json传输) + +``` java +@Configuration +public class RabbitMqConfig { + + @Bean + public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactoryPlus( + SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory, + Jackson2JsonMessageConverter jackson2JsonMessageConverter) { + rabbitListenerContainerFactory.setMessageConverter(jackson2JsonMessageConverter); + return rabbitListenerContainerFactory; + } + + @Bean + public Jackson2JsonMessageConverter jackson2JsonMessageConverter(ObjectMapper xssObjectMapper) { + return new Jackson2JsonMessageConverter(xssObjectMapper); + } + +} +``` + +编写配置文件 + +``` properties +spring.rabbitmq.host=192.168.134.100 +spring.rabbitmq.port=5672 +spring.rabbitmq.username=dev_udf-sample +spring.rabbitmq.password=1qazxsw2 +spring.rabbitmq.virtual-host=/dev_udf-sample +spring.rabbitmq.template.retry.enabled=true #发送方是否重试 +spring.rabbitmq.listener.retry.enabled=true #消费方是否重试 +``` + +定义公共的消息类 + +``` java +public class RabbitmqMessage implements Serializable { + private static final long serialVersionUID = 1L; + //消息ID + private String id; + ....其他自定义 +} +``` + + +###2.发布方样例 + +创建Exchange,这里使用的是DirectExchange,exchange主要是定义路由规则的,还有其他不同的路由规则实现,如:TopicExchange,他们都继承至AbstractExchange + +``` java + @Bean + public DirectExchange testExchange() { + return new DirectExchange("test_exchange"); + } +``` + +使用AmqpTemplate发送异步消息(RoutingKey则是指定消息的路由键,不同的路由键可被不同的消费方消费) + +``` java + @Autowired + private AmqpTemplate amqpTemplate; + + //然后调用发送方法发送消息 + this.amqpTemplate.convertAndSend("test_exchange", "testRoutingKey", new RabbitmqMessage("test")); +``` + +###3.消费方样例 + +创建消费队列,死信队列,以及与exchange的绑定关系 + +``` java + //消费队列 + @Bean + public Queue testConsume() { + //死信exchange与上面的定义方式一样 + Map args = new HashMap<>(); + args.put("x-dead-letter-exchange","test_exchange_dlx"); + args.put("x-dead-letter-routing-key","testRoutingKey_dlx"); + return new Queue("test_consume", true, false, false, args); + } + + //死信消费队列 + @Bean + public Queue testConsumeDlx() { + return new Queue("test_consume_dlx"); + } + + //消费队列绑定 + @Bean + public Binding testConsumeBinding() { + return new Binding("test_consume", DestinationType.QUEUE, + "test_exchange","testRoutingKey", null); + } + + //死信消费队列绑定 + @Bean + public Binding testConsumeDlxBinding() { + return new Binding("test_consume_dlx", DestinationType.QUEUE, + "test_exchange_dlx","testRoutingKey_dlx", null); + } +``` + +消费消息 + +``` java + @RabbitListener(queues = "test_consume") + public void process(Message message) { + log.info(message); + } +``` + + +##结束 + +以上演示了rabbitmq在spring boot中的配置,以及发送方和消费方的样例,在后续的章节中,会找机会介绍rabbitmq的搭建以及配置. + +--------- + +想获得最快更新,请关注公众号 + +![想获得最快更新,请关注公众号](https://mmbiz.qlogo.cn/mmbiz_jpg/gjOvoY7GOt5a4dicfGbqze591YAEiaRONE0nOsiaur4nlsmKtUpRuONue28wJ9JfOXfBl99OoVmYncohMnEY4LMdg/0?wx_fmt=jpeg "想获得最快更新,请关注公众号") +