前言
rabbitmq是一种我们经常使用的消息中间件，rabbitmq是实现amqp（高级消息队列协议）的消息中间件的一种，最初起源于金融系统，用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。rabbitmq主要是为了实现系统之间的双向解耦而实现的。当生产者大量产生数据时，消费者无法快速消费，那么需要一个中间层。保存这个数据。
amqp，即advanced message queuing protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。amqp的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。
rabbitmq是一个开源的amqp实现，服务器端用erlang语言编写，支持多种客户端，如：python、ruby、.net、java、jms、c、php、actionscript、xmpp、stomp等，支持ajax。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。
今天这篇，我们来看看spring boot是如何集成rabbitmq，发送消息和消费消息的。同时我们介绍下死信队列。
集成rabbitmq
集成rabbitmq只需要如下几步即可
1、添加maven依赖
|
1
2
3
4
5
6
7
8
9
|
<!--rabbitmq--><dependency> <groupid>org.springframework.boot</groupid> <artifactid>spring-boot-starter-amqp</artifactid></dependency> |
2、添加配置文件application.yaml
在application.yaml添加配置内容如下
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
|
spring: rabbitmq: host: 192.168.1.161 port: 5672 username: guest password: guest cache: channel: size: 10 listener: type: simple simple: acknowledge-mode: auto concurrency: 5 default-requeue-rejected: true max-concurrency: 100 retry: enabled: true # initial-interval: 1000ms max-attempts: 3 # max-interval: 1000ms multiplier: 1 stateless: true # publisher-confirms: true</pre> |
注意：
这里最基本的配置只需要配置host,port,username和password四个属性即可
其他属性都有各自的含义，比如retry是用于配置重试策略的，acknowledge-mode是配置消息接收确认机制的。
3、编写配置类
编写rabbitconfig配置类，采用java configuration的方式配置rabbittemplate、exchange和queue等信息，具体如下所示
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
|
package com.jackie.springbootdemo.config;import org.springframework.amqp.core.*;import org.springframework.amqp.rabbit.config.simplerabbitlistenercontainerfactory;import org.springframework.amqp.rabbit.connection.connectionfactory;import org.springframework.amqp.rabbit.core.rabbittemplate;import org.springframework.amqp.support.converter.jackson2jsonmessageconverter;import org.springframework.beans.factory.initializingbean;import org.springframework.beans.factory.annotation.autowired;import org.springframework.beans.factory.annotation.qualifier;import org.springframework.beans.factory.annotation.value;import org.springframework.beans.factory.config.configurablebeanfactory;import org.springframework.context.annotation.bean;import org.springframework.context.annotation.configuration;import org.springframework.context.annotation.scope;import java.util.hashmap;import java.util.map;@configuration public class rabbitmqconfig implements initializingbean { @autowired simplerabbitlistenercontainerfactory simplerabbitlistenercontainerfactory; @override public void afterpropertiesset() throws exception { simplerabbitlistenercontainerfactory.setmessageconverter(new jackson2jsonmessageconverter()); } @bean("jackson2jsonmessageconverter") public jackson2jsonmessageconverter jackson2jsonmessageconverter(connectionfactory connectionfactory) { return new jackson2jsonmessageconverter(); } @bean("rabbittemplate") @scope(configurablebeanfactory.scope_singleton) public rabbittemplate rabbittemplate(connectionfactory connectionfactory, @qualifier("jackson2jsonmessageconverter") jackson2jsonmessageconverter jackson2jsonmessageconverter) { rabbittemplate template = new rabbittemplate(connectionfactory); template.setmessageconverter(new jackson2jsonmessageconverter()); return template; } // --------------------- 声明队列 ------------------------ @bean public queue demoqueue() { return new queue("demo_queue"); } // --------------------- 声明exchange ------------------------ @bean public directexchange demoexchange() { return new directexchange("demo_exchange"); } // --------------------- 队列绑定 ------------------------ @bean public binding bindingalbumitemcreatedqueue(directexchange demoexchange, queue demoqueue) { return bindingbuilder.bind(demoqueue).to(demoexchange).with("100"); } } |
注意
这里声明了direct模式的exchange，声明一个queue，并通过routing-key为100将demo_queue绑定到demo_exchange，这样demo_queue就可以接收到demo_exchange发送的消息了。
4、编写消息发送类
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
|
package com.jackie.springbootdemo.message;import org.springframework.amqp.rabbit.core.rabbittemplate;import org.springframework.amqp.rabbit.support.correlationdata;import org.springframework.beans.factory.annotation.autowired;import org.springframework.stereotype.component;@component public class sender implements rabbittemplate.confirmcallback { private rabbittemplate rabbittemplate; /** * 构造方法注入 */ @autowired public sender(rabbittemplate rabbittemplate) { this.rabbittemplate = rabbittemplate; rabbittemplate.setconfirmcallback(this); //rabbittemplate如果为单例的话，那回调就是最后设置的内容 } public void sendmsg(string content) { rabbittemplate.convertandsend("demo_exchange", "100", content); } /** * 回调 */ @override public void confirm(correlationdata correlationdata, boolean ack, string cause) { system.out.println(" 回调id:" + correlationdata); if (ack) { system.out.println("消息成功消费"); } else { system.out.println("消息消费失败:" + cause); } } } |
注意
发送内容content，路由到routing-key为100上，则我们就可以在demo_queue队列中看到发送的消息内容了
confirm函数是回调函数，这里因为没有消费者，且acknoledge-mode是auto(其他两种值分别是none和manual)，所以ack是false。
5、编写发送消息测试类
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
|
package com.jackie.springbootdemo;import com.jackie.springbootdemo.message.sender;import org.junit.test;import org.junit.runner.runwith;import org.springframework.beans.factory.annotation.autowired;import org.springframework.boot.test.context.springboottest;import org.springframework.test.context.junit4.springrunner;import org.springframework.test.context.web.webappconfiguration;@runwith(springrunner.class) @springboottest(classes = springbootdemoapplication.class) @webappconfiguration public class rabbitapplicationtests { @autowired sender sender; @test public void contextloads() throws exception { sender.sendmsg("test"); } } |
运行该测试类，我们可以看到如下结果
6、编写消息消费类
|
1
2
3
4
5
6
7
8
9
|
package com.jackie.springbootdemo.message;import org.springframework.amqp.rabbit.annotation.rabbitlistener;import org.springframework.stereotype.component;@component public class receiver { @rabbitlistener(queues = "demo_queue") public void created(string message) { system.out.println("orignal message: " + message); } } |
注意
消息消费类也非常简单，添加注解@rabbitlistener，指定要监听的队列名称即可
除了注解@rabbitlistener，我们经常还能看到@rabbithandler，这两个注解可以配合起来使用。
@rabbitlistener 标注在类上面表示当有收到消息的时候，就交给 @rabbithandler 的方法处理，具体使用哪个方法处理，根据 messageconverter 转换后的参数类型，形如
|
1
2
3
4
5
6
|
@rabbitlistener(queues = "demo_queue") public class receiver { @rabbithandler public void processmessage1(string message) { system.out.println(message); } @rabbithandler public void processmessage2(byte[] message) { system.out.println(new string(message)); } } |
7、运行消息发送测试类
从执行结果可以看到，因为有了消费者，所以这次打印的结果是"消息消费成功"
而且，我们看到receiver类将消息消费并打印出消息的内容为"test"。
代码已经提交至项目rome：https://github.com/dminerjackie/rome
总结
以上就是这篇文章的全部内容了，希望本文的内容对大家的学习或者工作具有一定的参考学习价值，如果有疑问大家可以留言交流，谢谢大家对服务器之家的支持。
原文链接：https://www.cnblogs.com/bigdataZJ/p/springboot-rabbitmq.html
