服务器之家：专注于VPS、云服务器配置技术及软件下载分享
分类导航

PHP教程|ASP.NET教程|Java教程|ASP教程|编程技术|正则表达式|C/C++|IOS|C#|Swift|Android|VB|R语言|JavaScript|易语言|vb.net|

服务器之家 - 编程语言 - Java教程 - SpringBoot之RabbitMQ的使用方法

SpringBoot之RabbitMQ的使用方法

2021-06-24 09:29小崔的笔记本 Java教程

这篇文章主要介绍了SpringBoot之RabbitMQ的使用方法，详细的介绍了2种模式，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧

一 、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

    • 延伸 · 阅读

    精彩推荐
    最近更新
    编辑推荐
    阅读排行
    热门标签
    501 　 406 　 415 　 2000 　 1433 　 401 　 330 　 htmlparser 　 选择排序法 　 随机抽取 　 内存卡 　 下载器 　 十六进制字符 　 响应 　 静态工厂 　 构造函数 　 swt 　 Overview 　 对象池模式 　 子线程 　 支票 　 金额大写转换 　 空心菱形 　 AspectJ 　 面向切面编程 　 方法名 　 网址信息 　 动态代理模式 　 线程安全 　 ZGC 　
    © 2019-2022 服务器之家(www.tuohang.net) 版权所有关于我们联系我们版权申明网站地图