服务器之家

服务器之家 > 正文

Spring Boot系列教程之7步集成RabbitMQ的方法

时间:2021-06-11 13:44     来源/作者:JackieZheng

前言

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");
 } }

运行该测试类,我们可以看到如下结果

Spring Boot系列教程之7步集成RabbitMQ的方法

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、运行消息发送测试类

Spring Boot系列教程之7步集成RabbitMQ的方法

从执行结果可以看到,因为有了消费者,所以这次打印的结果是"消息消费成功"

而且,我们看到receiver类将消息消费并打印出消息的内容为"test"。

代码已经提交至项目rome:https://github.com/dminerjackie/rome

总结

以上就是这篇文章的全部内容了,希望本文的内容对大家的学习或者工作具有一定的参考学习价值,如果有疑问大家可以留言交流,谢谢大家对服务器之家的支持。

原文链接:https://www.cnblogs.com/bigdataZJ/p/springboot-rabbitmq.html

相关文章

热门资讯

2020微信伤感网名听哭了 让对方看到心疼的伤感网名大全
2020微信伤感网名听哭了 让对方看到心疼的伤感网名大全 2019-12-26
yue是什么意思 网络流行语yue了是什么梗
yue是什么意思 网络流行语yue了是什么梗 2020-10-11
背刺什么意思 网络词语背刺是什么梗
背刺什么意思 网络词语背刺是什么梗 2020-05-22
苹果12mini价格表官网报价 iPhone12mini全版本价格汇总
苹果12mini价格表官网报价 iPhone12mini全版本价格汇总 2020-11-13
2021德云社封箱演出完整版 2021年德云社封箱演出在线看
2021德云社封箱演出完整版 2021年德云社封箱演出在线看 2021-03-15
返回顶部