服务器之家

服务器之家 > 正文

SpringBoot之RabbitMQ的使用方法

时间：2021-06-24 09:29     来源/作者：小崔的笔记本

一 、rabbitmq的介绍

rabbitmq是消息中间件的一种，消息中间件即分布式系统中完成消息的发送和接收的基础软件，消息中间件的工作过程可以用生产者消费者模型来表示.即,生产者不断的向消息队列发送信息，而消费者从消息队列中消费信息.具体过程如下：

SpringBoot之RabbitMQ的使用方法

从上图可看出，对于消息队列来说，生产者、消息队列、消费者是最重要的三个概念，生产者发消息到消息队列中去，消费者监听指定的消息队列，并且当消息队列收到消息之后，接收消息队列传来的消息，并且给予相应的处理。消息队列常用于分布式系统之间互相信息的传递。

对于rabbitmq来说，除了这三个基本模块以外，还添加了一个模块,即交换机(exchange)。它使得生产者和消息队列之间产生了隔离，生产者将消息发送给交换机，而交换机则根据调度策略把相应的消息转发给对应的消息队列。

交换机的主要作用是接收相应的消息并且绑定到指定的队列。交换机有四种类型，分别为direct、topic、headers、fanout。

direct是rabbitmq默认的交换机模式，也是最简单的模式。即创建消息队列的时候，指定一个bindingkey。当发送者发送消息的时候，指定对应的key。当key和消息队列的bindingkey一致的时候，消息将会被发送到该消息队列中。

topic转发信息主要是依据通配符，队列和交换机的绑定主要是依据一种模式(通配符+字符串)，而当发送消息的时候，只有指定的key和该模式相匹配的时候,消息才会被发送到该消息队列中。

headers也是根据一个规则进行匹配，在消息队列和交换机绑定的时候会指定一组键值对规则，而发送消息的时候也会指定一组键值对规则，当两组键值对规则相匹配的时候，消息会被发送到匹配的消息队列中。

fanout是路由广播的形式，将会把消息发给绑定它的全部队列，即便设置了key，也会被忽略。

二 、springboot整合rabbitmq(direct模式)

springboot整合rabbitmq非常简单，首先还是pom.xml引入依赖。

?
1
2
3
4
<dependency>
     <groupid>org.springframework.boot</groupid>
     <artifactid>spring-boot-starter-amqp</artifactid>
 </dependency>

在application.properties中配置rabbitmq相关的信息，并首先启动了rabbitmq实例，并创建两个queue。

SpringBoot之RabbitMQ的使用方法

?
1
2
3
4
5
spring.application.name=spirng-boot-rabbitmq
spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=admin

配置queue(消息队列)，由于采用的是direct模式，需要在配置queue的时候指定一个键，使其和交换机绑定。

?
1
2
3
4
5
6
7
8
9
10
@configuration
public class rabbitconfig {
  
  @bean
  public org.springframework.amqp.core.queue queue() {
 
    return new org.springframework.amqp.core.queue("hello");
 
  }
}

接着就可以发送消息啦。在springboot中，我们使用amqptemplate去发送消息。代码如下：

?
1
2
3
4
5
6
7
8
9
10
11
12
13
@component
public class hellosender {
  
  @autowired
  private amqptemplate rabbittemplate;
 
  public void send(int index) {
 
    string context = "hello queue "+index + new date();
    system.out.println("sender : " + context);
    this.rabbittemplate.convertandsend("hello", context);
  }
}

生产者发送消息之后就需要消费者接收消息。这里定义了两个消息消费者，用来模拟生产者与消费者一对多的关系。

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
@component
@rabbitlistener(queues = "hello")
public class helloreceiver {
  
  @rabbithandler
  public void process(string hello) {
    system.out.println("receiver1 : " + hello);
  }
}
@component
@rabbitlistener(queues = "hello")
public class helloreceiver2 {
  
  @rabbithandler
  public void process(string hello) {
    system.out.println("receiver2 : " + hello);
  }
}

在单元测试中模拟发送消息，批量发送10条消息，两个接收者分别接收了5条消息。

?
1
2
3
4
5
6
7
8
9
@autowired
  private hellosender hellosender;
  @test
  public void hello() throws exception {
    for(int i=0;i<10;i++)
    {
      hellosender.send(i);
    }
  }

SpringBoot之RabbitMQ的使用方法

实际上rabbitmq还可以支持发送对象，当然由于涉及到序列化和反序列化，该对象要实现serilizable接口。这里定义了user对象，用来做发送消息内容。

?
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
import java.io.serializable;
public class user implements serializable{
  
  private string name;
  private string pwd;
  
  public string getpwd() {
    return pwd;
  }
  public void setpwd(string pwd) {
    this.pwd = pwd;
  }
  public string getname() {
    return name;
  }
  public void setname(string name) {
    this.name = name;
  }
  public user(string name, string pwd) {
    this.name = name;
    this.pwd = pwd;
  } 
  @override
public string tostring() {
    return "user{" +"name='" + name + '\'' +", pwd='" + pwd + '\'' +'}';
  }
}

在生产者中发送user对象。

?
1
2
3
4
5
6
7
8
9
10
11
12
13
@component
public class modelsender {
  
  @autowired
  private amqptemplate rabbittemplate;
  
  public void sendmodel(user user) {
 
    system.out.println("sender object: " + user.tostring());
    this.rabbittemplate.convertandsend("object", user);
 
  }
}

在消费者中接收user对象。

?
1
2
3
4
5
6
7
8
9
10
11
@component
@rabbitlistener(queues = "object")
public class modelrecevicer {
  
  @rabbithandler
  public void process(user user) {
 
    system.out.println("receiver object : " + user);
 
  }
}

在单元测试中注入modelsender 对象，实例化user对象，然后发送。

?
1
2
3
4
5
6
7
8
@autowired
  private modelsender modelsender;
  @test
  public void model() throws exception {
    
    user user=new user("abc","123");   
    modelsender.sendmodel(user);
  }

SpringBoot之RabbitMQ的使用方法

三 、springboot整合rabbitmq(topic转发模式)

首先需要在rabbitmq服务端创建交换机topicexchange，并绑定两个queue：topic.message、topic.messages。

SpringBoot之RabbitMQ的使用方法

新建topicrabbitconfig,设置对应的queue与binding。

?
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
@configuration
public class topicrabbitconfig {
  
  final static string message = "topic.message";
  final static string messages = "topic.messages";
  
  @bean
  public queue queuemessage() {
    return new queue(topicrabbitconfig.message);
  }
  
  @bean
  public queue queuemessages() {
    return new queue(topicrabbitconfig.messages);
  }
 
  @bean
  topicexchange exchange() {
    return new topicexchange("topicexchange");
  }
 
  @bean
  binding bindingexchangemessage(queue queuemessage, topicexchange exchange) {
    return bindingbuilder.bind(queuemessage).to(exchange).with("topic.message");
  }
 
  @bean
  binding bindingexchangemessages(queue queuemessages, topicexchange exchange) {
    return bindingbuilder.bind(queuemessages).to(exchange).with("topic.#");
  }
}

创建消息生产者，在topicsender中发送3个消息。

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
@component
public class topicsender {
  
  @autowired
  private amqptemplate rabbittemplate;
 
  public void send() {
    string context = "hi, i am message all";
    system.out.println("sender : " + context);
    this.rabbittemplate.convertandsend("topicexchange", "topic.1", context);
  }
 
  public void send1() {
    string context = "hi, i am message 1";
    system.out.println("sender : " + context);
    this.rabbittemplate.convertandsend("topicexchange", "topic.message", context);
  }
 
  public void send2() {   
    string context = "hi, i am messages 2";
    system.out.println("sender : " + context);
    this.rabbittemplate.convertandsend("topicexchange", "topic.messages", context);   
  }
}

生产者发送消息，这里创建了两个接收消息的消费者。

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
@component
@rabbitlistener(queues = "topic.message")
public class topicreceiver {
  
  @rabbithandler
  public void process(string message) {
    system.out.println("topic receiver1 : " + message);
  }
}
 
@component
@rabbitlistener(queues = "topic.messages")
public class topicreceiver2 {
  
  @rabbithandler
  public void process(string message) {
 
    system.out.println("topic receiver2 : " + message);
 
  }
}

在单元测试中注入topicsender，利用topicsender 发送消息。

?
1
2
3
4
5
6
7
8
@autowired
  private topicsender topicsender;
  @test
  public void topicsender() throws exception {     
    topicsender.send();   
    topicsender.send1();   
    topicsender.send2();
  }

SpringBoot之RabbitMQ的使用方法

从上面的输出结果可以看到，topic receiver2 匹配到了所有消息，topic receiver1只匹配到了1个消息。

四 、springboot整合rabbitmq(fanout exchange形式)

fanout exchange形式又叫广播形式，因此我们发送到路由器的消息会使得绑定到该路由器的每一个queue接收到消息。首先需要在rabbitmq服务端创建交换机fanoutexchange，并绑定三个queue：fanout.a、fanout.b、fanout.c。

SpringBoot之RabbitMQ的使用方法

与topic类似，新建fanoutrabbitconfig，绑定交换机和队列。

?
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
@configuration
public class fanoutrabbitconfig {
  
  @bean
  public queue amessage() {
    return new queue("fanout.a");
  }
  
  @bean
  public queue bmessage() {
    return new queue("fanout.b");
  }
 
  @bean
  public queue cmessage() {
    return new queue("fanout.c");
  }
 
  @bean
  fanoutexchange fanoutexchange() {
    return new fanoutexchange("fanoutexchange");
  }
 
  @bean
  binding bindingexchangea(queue amessage,fanoutexchange fanoutexchange) {
    return bindingbuilder.bind(amessage).to(fanoutexchange);
  }
 
  @bean
  binding bindingexchangeb(queue bmessage, fanoutexchange fanoutexchange) {
    return bindingbuilder.bind(bmessage).to(fanoutexchange);
  }
 
  @bean
  binding bindingexchangec(queue cmessage, fanoutexchange fanoutexchange) {
    return bindingbuilder.bind(cmessage).to(fanoutexchange);
  }
}

创建消息生产者，在fanoutsender中发送消息。

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
@component
public class fanoutsender {
 
  @autowired
  private amqptemplate rabbittemplate;
 
  public void send() {
    
    string context = "hi, fanout msg ";
    system.out.println("fanoutsender : " + context);
    this.rabbittemplate.convertandsend("fanoutexchange","", context);
 
  }
}

然后创建了3个接收者fanoutreceivera、fanoutreceiverb、fanoutreceiverc。

?
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
@component
@rabbitlistener(queues = "fanout.a")
public class fanoutreceivera {
  
  @rabbithandler
  public void process(string message) {
    system.out.println("fanout receiver a : " + message);
  }
}
@component
@rabbitlistener(queues = "fanout.b")
public class fanoutreceiverb {
  
  @rabbithandler
  public void process(string message) {
    system.out.println("fanout receiver b: " + message);
  }
}
@component
@rabbitlistener(queues = "fanout.c")
public class fanoutreceiverc {
  
  @rabbithandler
  public void process(string message) {
    system.out.println("fanout receiver c: " + message);
  }
}

在单元测试中注入消息发送者，发送消息。

?
1
2
3
4
5
6
@autowired
 private fanoutsender fanoutsender;
 @test
 public void fanoutsender() throws exception {
   fanoutsender.send();
 }

从下图可以看到3个队列都接收到了消息。

SpringBoot之RabbitMQ的使用方法

本章节创建的类比较多，下图为本章节的结构，也可以直接查看demo源码了解。

SpringBoot之RabbitMQ的使用方法

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持服务器之家。

原文链接：http://www.cnblogs.com/5ishare/p/10163318.html

标签：
springboot rabbitmq 

相关文章

热门资讯

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
返回顶部