rabbitmq-交换机
1、交换机是干什么的？
消息(message)由client发送，rabbitmq接收到消息之后通过交换机转发到对应的队列上面。worker会从队列中获取未被读取的数据处理。
1、交换机的种类
rabbitmq包含四种不同的交换机类型：
- direct exchange：直连交换机，转发消息到routigkey指定的队列
- fanout exchange：扇形交换机，转发消息到所有绑定队列（速度最快）
- topic exchange：主题交换机，按规则转发消息（最灵活）
- headers exchange：首部交换机 （不常用）
1、direct exchange（直连交换机）
直连交换机是一种带路由功能的交换机，根据消息携带的路由键将消息投递给对应队列。
一个队列通过routing_key（路由键）与一个交换机绑定，当消息被发送的时候，需要指定一个routing_key，这个消息被送达交换机的时候，就会被交换机送到指定的队列里面去。
类似一对一的关系！
1.1 springboot的简单实现
创建2个springboot项目，一个 rabbitmq-provider （生产者），一个rabbitmq-consumer（消费者）。
1、首先创建 rabbitmq-provider
pom.xml添加依赖：
|
1
2
3
4
5
|
<!--rabbitmq--> <dependency> <groupid>org.springframework.boot</groupid> <artifactid>spring-boot-starter-amqp</artifactid> </dependency> |
配置application.yml：
|
1
2
3
4
5
6
7
8
9
10
|
server: port: 8090spring: application: name: rabbitmq-provider rabbitmq: host: 192.168.152.173 port: 5672 #注意15672是web页面端口，消息发送端口为5672 username: guest password: guest |
编写配置文件
|
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
|
@configurationpublic class directrabbitconfig { /** * 创建消息队列 起名：testdirectqueue * durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效 * exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable * autodelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。 * 一般设置一下队列的持久化就好,其余两个就是默认false * @return new queue(name,durable,exclusive,autodelete) */ @bean public queue testdirectqueue() { return new queue("testdirectqueue",true); } /** * direct交换机 起名：testdirectexchange */ @bean directexchange testdirectexchange() { return new directexchange("testdirectexchange",true,false); } /** * 绑定 * 将队列和交换机绑定, 并设置用于匹配键（路由键）：testdirectrouting */ @bean binding bindingdirect() { return bindingbuilder.bind(testdirectqueue()).to(testdirectexchange()).with("testdirectrouting"); } } |
编写controller层
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
|
@restcontroller@requestmapping("/rabbit")public class sendmessagecontroller { @autowired private rabbittemplate rabbittemplate; //使用rabbittemplate,这提供了接收/发送等等方法 @requestmapping("/send") public string senddirectmessage() { string messageid = string.valueof(uuid.randomuuid()); string messagedata = "league of legends never dies！"; string createtime = localdatetime.now().format(datetimeformatter.ofpattern("yyyy-mm-dd hh:mm:ss")); map<string,object> map=new hashmap<>(); map.put("messageid",messageid); map.put("messagedata",messagedata); map.put("createtime",createtime); //将消息携带绑定键值：testdirectrouting 发送到交换机testdirectexchange rabbittemplate.convertandsend("testdirectexchange", "testdirectrouting", map); return "ok"; }} |
把rabbitmq-provider项目运行，调用下接口:
因为我们目前还没弄消费者 rabbitmq-consumer，消息没有被消费的，我们去rabbitmq管理页面看看，是否推送成功：
消息已经推送到rabbitmq服务器上面了。
2、创建rabbitmq-consumer项目：
pom.xml和application.yml跟上边一样（把端口号和应用名换了即可）
创建消息接收监听类，directreceiver01.java：
|
1
2
3
4
5
6
7
8
|
@component@rabbitlistener(queues = "testdirectqueue")//监听的队列名称 testdirectqueuepublic class directreceiver01 { @rabbithandler public void process(map testmessage) { system.out.println("directreceiver01消费者收到的消息: " + testmessage.tostring()); }} |
启动消费者项目：
运行结果如下
两个消息直接被消费了。然后可以再继续调用rabbitmq-provider项目的推送消息接口，可以看到消费者即时消费消息！
扩展
直连交换机既然是一对一，那如果咱们配置多台监听绑定到同一个直连交互的同一个队列，会怎么样？
直接复制两份directreceiver.java，改下打印内容！
重新启动项目！在生产者多发送几次消息。运行结果如下：
可以看到是实现了轮询的方式对消息进行消费，而且不存在重复消费。
2、topic exchange（主题交换机）
直连交换机的routing_key方案非常简单，但是它是一对一的关系，那么我们需要一对多呢？希望一条消息发送给多个队列。所以rabbitmq提供了一种主题交换机，发送到主题交换机上的消息需要携带指定规则的routing_key，主题交换机会根据这个规则将数据发送到对应的(多个)队列上。
主题交换机的routing_key需要有一定的规则，交换机和队列的binding_key需要采用 *.# 的格式，每个部分用.分开，其中：
*表示一个单词
#表示任意数量（零个或多个）单词。（例如：topic.man）
1、在rabbitmq-provider项目里面创建topicrabbitconfig.java：
|
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
|
@configurationpublic class topicrabbitconfig { //绑定键 public final static string man = "topic.man"; public final static string woman = "topic.woman"; @bean public queue firstqueue() { return new queue(topicrabbitconfig.man); } @bean public queue secondqueue() { return new queue(topicrabbitconfig.woman); } @bean topicexchange exchange() { return new topicexchange("topicexchange"); } //将firstqueue和topicexchange绑定,而且绑定的键值为topic.man //这样只要是消息携带的路由键是topic.man,才会分发到该队列 @bean binding bindingexchangemessage() { return bindingbuilder.bind(firstqueue()).to(exchange()).with(man); } //将secondqueue和topicexchange绑定,而且绑定的键值为用上通配路由键规则topic.# // 这样只要是消息携带的路由键是以topic.开头,都会分发到该队列 @bean binding bindingexchangemessage2() { return bindingbuilder.bind(secondqueue()).to(exchange()).with("topic.#"); }} |
编写controller层
|
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
|
@requestmapping("rabbit")@restcontrollerpublic class topicsendcontroller { @autowired private rabbittemplate template; @getmapping("/sendtopicmessage1") public string sendtopicmessage1() { string messageid = string.valueof(uuid.randomuuid()); string messagedata = "message: m a n "; string createtime = localdatetime.now().format(datetimeformatter.ofpattern("yyyy-mm-dd hh:mm:ss")); map<string, object> manmap = new hashmap<>(); manmap.put("messageid", messageid); manmap.put("messagedata", messagedata); manmap.put("createtime", createtime); template.convertandsend("topicexchange", "topic.man", manmap); return "ok"; } @getmapping("/sendtopicmessage2") public string sendtopicmessage2() { string messageid = string.valueof(uuid.randomuuid()); string messagedata = "message: woman is all "; string createtime = localdatetime.now().format(datetimeformatter.ofpattern("yyyy-mm-dd hh:mm:ss")); map<string, object> womanmap = new hashmap<>(); womanmap.put("messageid", messageid); womanmap.put("messagedata", messagedata); womanmap.put("createtime", createtime); template.convertandsend("topicexchange", "topic.woman", womanmap); return "ok"; }} |
不要急着去运行，先去编写消费者,创建两个接口
topicmanreceiver :
|
1
2
3
4
5
6
7
8
|
@component@rabbitlistener(queues = "topic.man")public class topicmanreceiver { @rabbithandler public void process(map testmessage) { system.out.println("topicmanreceiver消费者收到消息 : " + testmessage.tostring()); }} |
topicwomanreceiver :
|
1
2
3
4
5
6
7
8
|
@component@rabbitlistener(queues = "topic.woman")public class topicwomanreceiver { @rabbithandler public void process(map testmessage) { system.out.println("topicwomanreceiver消费者收到消息 : " + testmessage.tostring()); }} |
然后把rabbitmq-provider，rabbitmq-consumer两个项目都跑起来，先调用/sendtopicmessage1 接口：
然后看消费者rabbitmq-consumer的控制台输出情况：
topicmanreceiver监听队列1，绑定键为：topic.man
topictotalreceiver监听队列2，绑定键为：topic.#
而当前推送的消息，携带的路由键为：topic.man
所以可以看到两个监听消费者receiver都成功消费到了消息，因为这两个recevier监听的队列的绑定键都能与这条消息携带的路由键匹配上。
用/sendtopicmessage2接口：
然后看消费者rabbitmq-consumer的控制台输出情况：
topicmanreceiver监听队列1，绑定键为：topic.man
topictotalreceiver监听队列2，绑定键为：topic.#
而当前推送的消息，携带的路由键为：topic.woman
所以可以看到两个监听消费者，只有topicwomanreceiver 成功消费到了消息，
3、fanout exchange（扇形交换机）
扇形交换机是最基本的交换机类型，它做的事情很简单–广播信息。fanout交换机会把接收到的消息全部转发到绑定的队列上。因为广播不需要“思考”，所以fanout交换机是四种交换机中速度最快的。
同样地，先在rabbitmq-provider项目上创建fanoutrabbitconfig.java：
|
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
|
@configurationpublic class fanoutrabbitconfig { /** * 创建三个队列 ：fanout.a fanout.b fanout.c * 将三个队列都绑定在交换机 fanoutexchange 上 * 因为是扇型交换机, 路由键无需配置,配置也不起作用 */ @bean public queue queuea() { return new queue("fanout.a"); } @bean public queue queueb() { return new queue("fanout.b"); } @bean public queue queuec() { return new queue("fanout.c"); } @bean fanoutexchange fanoutexchange() { return new fanoutexchange("fanoutexchange"); } @bean binding bindingexchangea() { return bindingbuilder.bind(queuea()).to(fanoutexchange()); } @bean binding bindingexchangeb() { return bindingbuilder.bind(queueb()).to(fanoutexchange()); } @bean binding bindingexchangec() { return bindingbuilder.bind(queuec()).to(fanoutexchange()); }} |
编写controller层：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
|
@restcontroller@requestmapping("rabbit")public class fanoutcontroller { @autowired private rabbittemplate template; @getmapping("/sendfanoutmessage") public string sendfanoutmessage() { string messageid = string.valueof(uuid.randomuuid()); string messagedata = "message: testfanoutmessage "; string createtime = localdatetime.now().format(datetimeformatter.ofpattern("yyyy-mm-dd hh:mm:ss")); map<string, object> map = new hashmap<>(); map.put("messageid", messageid); map.put("messagedata", messagedata); map.put("createtime", createtime); template.convertandsend("fanoutexchange", null, map); return "ok"; }} |
接着在rabbitmq-consumer项目里加上消息消费类，
fanoutreceivera:
|
1
2
3
4
5
6
7
8
|
@component@rabbitlistener(queues = "fanout.a")public class fanoutreceivera { @rabbithandler public void process(map testmessage) { system.out.println("fanoutreceiverc消费者收到消息 : " +testmessage.tostring()); }} |
fanoutreceiverb
|
1
2
3
4
5
6
7
8
|
@component@rabbitlistener(queues = "fanout.b")public class fanoutreceiverb{ @rabbithandler public void process(map testmessage) { system.out.println("fanoutreceiverc消费者收到消息 : " +testmessage.tostring()); }} |
fanoutreceiverc :
|
1
2
3
4
5
6
7
8
|
@component@rabbitlistener(queues = "fanout.c")public class fanoutreceiverc { @rabbithandler public void process(map testmessage) { system.out.println("fanoutreceiverc消费者收到消息 : " +testmessage.tostring()); }} |
最后将rabbitmq-provider和rabbitmq-consumer项目都跑起来，调用下接口/sendfanoutmessage，运行结果如下：
可以看到只要发送到 fanoutexchange 这个扇型交换机的消息， 三个队列都绑定这个交换机，所以三个消息接收类都监听到了这条消息。
4、消息确认（发送确认与接收确认） 4.1发送确认
在rabbitmq-provider项目中
编写配置文件：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
server: port: 8090spring: application: name: rabbitmq-consumer rabbitmq: host: 192.168.152.173 port: 5672 username: guest password: guest #确认消息已发送到交换机(exchange) publisher-confirm-type: correlated #确认消息已发送到队列(queue) publisher-returns: true |
配置相关的消息确认回调函数，rabbitconfig.java：
|
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
|
/** * 配置相关的消息确认回调函数 */@configurationpublic class rabbitconfig { @bean public rabbittemplate createrabbittemplate(connectionfactory connectionfactory) { rabbittemplate rabbittemplate = new rabbittemplate(); rabbittemplate.setconnectionfactory(connectionfactory); //设置开启mandatory,才能触发回调函数,无论消息推送结果怎么样都强制调用回调函数 rabbittemplate.setmandatory(true); //确认回调 rabbittemplate.setconfirmcallback(new rabbittemplate.confirmcallback() { @override public void confirm(correlationdata correlationdata, boolean condition, string cause) { system.out.println("confirmcallback: " + "相关数据：" + correlationdata); system.out.println("confirmcallback: " + "确认情况：" + condition); system.out.println("confirmcallback: " + "原因：" + cause); } }); //设置返回消息的回调 rabbittemplate.setreturncallback(new rabbittemplate.returncallback() { @override public void returnedmessage(message message, int replycode, string replytext, string exchange, string routingkey) { system.out.println("returncallback: " + "消息：" + message); system.out.println("returncallback: " + "回应码：" + replycode); system.out.println("returncallback: " + "回应信息：" + replytext); system.out.println("returncallback: " + "交换机：" + exchange); system.out.println("returncallback: " + "路由键：" + routingkey); } }); return rabbittemplate; }} |
confirmcallback
confirmcallback是一个回调接口，就是只确认是否正确到达 exchange 中。
returncallback
通过实现 returncallback 接口，启动消息失败返回，此接口是在交换器路由不到队列时触发回调，该方法可以不使用，因为交换器和队列是在代码里绑定的，如果存在绑定队列失败，那就是你代码写错了。
从总体的情况分析，推送消息存在四种情况：
1、消息推送到server，但是在server里找不到交换机
2、消息推送到server，找到交换机了，但是没找到队列
3、消息推送到sever，交换机和队列啥都没找到
4、消息推送成功
1、消息推送到server，但是在server里找不到交换机
写个测试接口，把消息推送到名为‘non-existent-exchange'的交换机上（这个交换机是没有创建没有配置的）：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
|
@getmapping("/testmessageack") public string testmessageack() { string messageid = string.valueof(uuid.randomuuid()); string messagedata = "message: non-existent-exchange test message "; string createtime = localdatetime.now().format(datetimeformatter.ofpattern("yyyy-mm-dd hh:mm:ss")); map<string, object> map = new hashmap<>(); map.put("messageid", messageid); map.put("messagedata", messagedata); map.put("createtime", createtime); //non-existent-exchange这个交换机没有创建和配置 template.convertandsend("non-existent-exchange", "testdirectrouting", map); return "ok"; } |
调用接口，查看rabbitmq-provuder项目的控制台输出情况
因为这个交换机没有创建，消息不能到达到达 exchange 中。所以会触发confirmcallback这个回调函数
结论： 只会触发confirmcallback。
2、消息推送到server，找到交换机了，但是没找到队列
新增一个交换机，但是不给这个交换机绑定队列，在directrabitconfig里面新增一个直连交换机，不做任何绑定配置操作：
|
1
2
3
4
|
@bean directexchange lonelydirectexchange() { return new directexchange("lonelydirectexchange"); } |
同样编写测试接口
|
1
2
3
4
5
6
7
8
9
10
11
12
|
@getmapping("/testmessageack2") public string testmessageack2() { string messageid = string.valueof(uuid.randomuuid()); string messagedata = "message: lonelydirectexchange test message "; string createtime = localdatetime.now().format(datetimeformatter.ofpattern("yyyy-mm-dd hh:mm:ss")); map<string, object> map = new hashmap<>(); map.put("messageid", messageid); map.put("messagedata", messagedata); map.put("createtime", createtime); rabbittemplate.convertandsend("lonelydirectexchange", "testdirectrouting", map); return "ok"; } |
调用方法,查看运行结果：
可以看到两个回调函数都被调用了；
这种情况下，消息是推送成功到服务器了的，所以confirmcallback对消息确认情况是true；
而在retruncallback回调函数的打印参数里面可以看到，消息是推送到了交换机成功了，但是在路由分发给队列的时候，找不到队列，所以报了错误 no_route。
结论： 两个回调函数都会触发！
3、消息推送到sever，交换机和队列啥都没找到
结论： 触发的是 confirmcallback 回调函数。
4、消息推送成功
结论： 触发的是 confirmcallback 回调函数。
总结： confirmcallback回调函数总是会被触发，确认情况true和false是有没有找到交换机，false的话打印原因。returncallback是交换机路由不到队列是回调，也就是你交换机没有绑定队列，这个回调函数可以不写，绑定失败的话就好好看看代码哪里写错了吧。
以上是生产者推送消息的消息确认 回调函数的使用介绍！
4.2 接受确认
消费者确认发生在监听队列的消费者处理业务失败，如，发生了异常，不符合要求的数据……，这些场景我们就需要手动处理，比如重新发送或者丢弃。
消息确认模式有：
- acknowledgemode.none：自动确认
- acknowledgemode.auto：根据情况确认
- acknowledgemode.manual：手动确认
参考：
4.1.1 自动确认
修改消费者配置文件
|
1
2
3
4
5
6
7
8
|
rabbitmq: host: 192.168.152.173 port: 5672 username: guest password: guest listener: direct: acknowledge-mode: auto # manual：手动 |
在消费者项目里，
新建messagelistenerconfig.java上添加代码相关的配置代码：
|
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
|
package com.zsn.demo.config;import com.zsn.demo.resceiver.myackreceiver;import org.springframework.amqp.core.acknowledgemode;import org.springframework.amqp.rabbit.connection.cachingconnectionfactory;import org.springframework.amqp.rabbit.listener.simplemessagelistenercontainer;import org.springframework.beans.factory.annotation.autowired;import org.springframework.context.annotation.bean;import org.springframework.context.annotation.configuration;/** * @author： zhouzhou * @date：2021/7/22 9:49 */@configurationpublic class messagelistenerconfig { @autowired private cachingconnectionfactory connectionfactory; @autowired private myackreceiver myackreceiver;//消息接收处理类 @bean public simplemessagelistenercontainer simplemessagelistenercontainer() { simplemessagelistenercontainer container = new simplemessagelistenercontainer(connectionfactory); container.setconcurrentconsumers(1); container.setmaxconcurrentconsumers(1); container.setacknowledgemode(acknowledgemode.manual); // rabbitmq默认是自动确认，这里改为手动确认消息 //设置一个队列 container.setqueuenames("testdirectqueue"); //如果同时设置多个如下： 前提是队列都是必须已经创建存在的 // container.setqueuenames("testdirectqueue","testdirectqueue2","testdirectqueue3"); //另一种设置队列的方法,如果使用这种情况,那么要设置多个,就使用addqueues //container.setqueues(new queue("testdirectqueue",true)); //container.addqueues(new queue("testdirectqueue2",true)); //container.addqueues(new queue("testdirectqueue3",true)); container.setmessagelistener(myackreceiver); return container; }} |
对应的手动确认消息监听类，myackreceiver.java（手动确认模式需要实现 channelawaremessagelistener）：
|
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
49
50
51
52
53
54
55
56
57
58
|
package com.zsn.demo.resceiver;import com.rabbitmq.client.channel;import org.springframework.amqp.core.message;import org.springframework.amqp.rabbit.listener.api.channelawaremessagelistener;import org.springframework.stereotype.component;import java.util.hashmap;import java.util.map;/** * @author： zhouzhou * @date：2021/7/22 9:50 */@componentpublic class myackreceiver implements channelawaremessagelistener { /** * 用于处理接收到的 rabbit 消息的回调。 * 实现者应该处理指定的信息 * 通常通过给定的会话发送回复消息 * @param message the received amqp message (never <code>null</code>) * @param channel the underlying rabbit channel (never <code>null</code>) * @throws exception any. */ @override public void onmessage(message message, channel channel) throws exception { long deliverytag = message.getmessageproperties().getdeliverytag(); try { //因为传递消息的时候用的map传递,所以将map从message内取出需要做些处理 string msg = message.tostring(); string[] msgarray = msg.split("'");//可以点进message里面看源码,单引号直接的数据就是我们的map消息数据 map<string, string> msgmap = mapstringtomap(msgarray[1].trim(),3); string messageid=msgmap.get("messageid"); string messagedata=msgmap.get("messagedata"); string createtime=msgmap.get("createtime"); system.out.println(" myackreceiver messageid:"+messageid+" messagedata:"+messagedata+" createtime:"+createtime); system.out.println("消费的主题消息来自："+message.getmessageproperties().getconsumerqueue()); channel.basicack(deliverytag, true); //第二个参数，手动确认可以被批处理，当该参数为 true 时，则可以一次性确认 delivery_tag 小于等于传入值的所有消息// channel.basicreject(deliverytag, true);//第二个参数，true会重新放回队列，所以需要自己根据业务逻辑判断什么时候使用拒绝 } catch (exception e) { channel.basicreject(deliverytag, false); e.printstacktrace(); } } //{key=value,key=value,key=value} 格式转换成map private map<string, string> mapstringtomap(string str,int entrynum ) { str = str.substring(1, str.length() - 1); string[] strs = str.split(",",entrynum); map<string, string> map = new hashmap<string, string>(); for (string string : strs) { string key = string.split("=")[0].trim(); string value = string.split("=")[1]; map.put(key, value); } return map; }} |
需要注意的 basicack 方法需要传递两个参数
- deliverytag（唯一标识 id）：当一个消费者向 rabbitmq 注册后，会建立起一个 channel ，rabbitmq 会用 basic.deliver 方法向消费者推送消息，这个方法携带了一个 delivery tag， 它代表了 rabbitmq 向该 channel 投递的这条消息的唯一标识 id，是一个单调递增的正整数，delivery tag 的范围仅限于 channel
- multiple：为了减少网络流量，手动确认可以被批处理，当该参数为 true 时，则可以一次性确认 delivery_tag 小于等于传入值的所有消息
basicnack方法需要传递三个参数
- deliverytag（唯一标识 id）：上面已经解释了。
- multiple：上面已经解释了。
- requeue： true ：重回队列，false ：丢弃，我们在nack方法中必须设置 false，否则重发没有意义。
basicreject方法需要传递两个参数
- deliverytag（唯一标识 id）：上面已经解释了。
- requeue：上面已经解释了，在reject方法里必须设置true。
总结
- basic.ack用于肯定确认
- basic.nack用于否定确认（注意：这是amqp 0-9-1的rabbitmq扩展）
- basic.reject用于否定确认，但与basic.nack相比有一个限制:一次只能拒绝单条消息
消费者端以上的3个方法都表示消息已经被正确投递，但是basic.ack表示消息已经被正确处理。
而basic.nack,basic.reject表示没有被正确处理：
这时，先调用接口，给直连交换机testdirectexchange 的队列testdirectqueue 推送一条消息，可以看到监听器正常消费了下来：
扩展1
除了直连交换机的队列testdirectqueue需要变成手动确认以外，我们还需要将一个其他的队列或者多个队列也变成手动确认，而且不同队列实现不同的业务处理。
那么我们需要做的第一步，往simplemessagelistenercontainer里添加多个队列
myackreceiver.java 就可以同时将上面设置到的队列的消息都消费下来。
|
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
49
50
51
52
53
54
55
56
57
|
@componentpublic class myackreceiver implements channelawaremessagelistener { /** * 用于处理接收到的 rabbit 消息的回调。 * 实现者应该处理指定的信息 * 通常通过给定的会话发送回复消息 * @param message the received amqp message (never <code>null</code>) * @param channel the underlying rabbit channel (never <code>null</code>) * @throws exception any. */ @override public void onmessage(message message, channel channel) throws exception { long deliverytag = message.getmessageproperties().getdeliverytag(); try { //因为传递消息的时候用的map传递,所以将map从message内取出需要做些处理 string msg = message.tostring(); string[] msgarray = msg.split("'");//可以点进message里面看源码,单引号直接的数据就是我们的map消息数据 map<string, string> msgmap = mapstringtomap(msgarray[1].trim(),3); string messageid=msgmap.get("messageid"); string messagedata=msgmap.get("messagedata"); string createtime=msgmap.get("createtime"); if ("testdirectqueue".equals(message.getmessageproperties().getconsumerqueue())){ system.out.println("消费的消息来自的队列名为："+message.getmessageproperties().getconsumerqueue()); system.out.println("消息成功消费到 messageid:"+messageid+" messagedata:"+messagedata+" createtime:"+createtime); system.out.println("执行testdirectqueue中的消息的业务处理流程......"); } if ("fanout.a".equals(message.getmessageproperties().getconsumerqueue())){ system.out.println("消费的消息来自的队列名为："+message.getmessageproperties().getconsumerqueue()); system.out.println("消息成功消费到 messageid:"+messageid+" messagedata:"+messagedata+" createtime:"+createtime); system.out.println("执行fanout.a中的消息的业务处理流程......"); } channel.basicack(deliverytag, true);// channel.basicreject(deliverytag, true);//为true会重新放回队列 } catch (exception e) { channel.basicreject(deliverytag, false); e.printstacktrace(); } } //{key=value,key=value,key=value} 格式转换成map private map<string, string> mapstringtomap(string str,int ennum) { str = str.substring(1, str.length() - 1); string[] strs = str.split(",",ennum); map<string, string> map = new hashmap<string, string>(); for (string string : strs) { string key = string.split("=")[0].trim(); string value = string.split("=")[1]; map.put(key, value); } return map; }} |
分别调用这两个队列相关接口，运行结果如下：
扩展2： 手动确认的另一种写法：
我们在实际工作中可能不会这么写，一般会写在server层，自己定义一个接口：
rabbitconsumerservice
|
1
2
3
|
public interface rabbitconsumerservice { void process(channel channel, message message) throws ioexception;} |
|
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
|
@componentpublic class rabbitconsumerserviceimpl implements rabbitconsumerservice{ @rabbithandler @rabbitlistener(queues = "testdirectqueue") @override public void process(channel channel, message message) throws ioexception { long deliverytag = message.getmessageproperties().getdeliverytag(); try { //因为传递消息的时候用的map传递,所以将map从message内取出需要做些处理 string msg = message.tostring(); string[] msgarray = msg.split("'");//可以点进message里面看源码,单引号直接的数据就是我们的map消息数据 map<string, string> msgmap = mapstringtomap(msgarray[1].trim(),3); string messageid=msgmap.get("messageid"); string messagedata=msgmap.get("messagedata"); string createtime=msgmap.get("createtime"); system.out.println(" myackreceiver messageid:"+messageid+" messagedata:"+messagedata+" createtime:"+createtime); system.out.println("消费的主题消息来自："+message.getmessageproperties().getconsumerqueue()); channel.basicack(deliverytag, true); //第二个参数，手动确认可以被批处理，当该参数为 true 时，则可以一次性确认 delivery_tag 小于等于传入值的所有消息// channel.basicreject(deliverytag, true);//第二个参数，true会重新放回队列，所以需要自己根据业务逻辑判断什么时候使用拒绝 } catch (exception e) { channel.basicreject(deliverytag, false); e.printstacktrace(); } } //{key=value,key=value,key=value} 格式转换成map private map<string, string> mapstringtomap(string str,int entrynum ) { str = str.substring(1, str.length() - 1); string[] strs = str.split(",",entrynum); map<string, string> map = new hashmap<string, string>(); for (string string : strs) { string key = string.split("=")[0].trim(); string value = string.split("=")[1]; map.put(key, value); } return map; }} |
@component这个理论上要换成@service，但是由于我没有添加日志，所以换成@service会报错，这里就是提供一个思路，运行结果如下：
附：
rabbitmq在application.yml文件中的其他配置
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
|
rabbitmq: addresses: 192.168.152.193:5672 username: guest password: guest #虚拟host 可以不设置,使用server默认host virtual-host: / listener: simple: prefetch: 1 #设置一次处理一个消息 acknowledge-mode: manual #设置消费端手动 ack concurrency: 3 #设置同时有3个消费者消费 connection-timeout: 500 # 确认消息是否正确到达queue,如果没有则触发，如果有则不触发 publisher-returns: true #确认消息已发送到交换机(exchange) publisher-confirm-type: correlated template: #设置为 true后 消费者在消息没有被路由到合适队列情况下会被return监听，而不会自动删除 mandatory: true |
解释：mandatory 设置为 true 则会监听器会接受到路由不可达的消息，然后处理。如果设置为 false，broker 将会自动删除该消息。
参考https://blog.csdn.net/qq_35387940/article/details/100514134
到此这篇关于rabbitmq交换机与springboot整合的简单实现的文章就介绍到这了,更多相关rabbitmq springboot整合内容请搜索服务器之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持服务器之家！
原文链接：https://blog.csdn.net/Pluto372/article/details/118966817
