SpringBoot之RabbitMQ的使用方法
一 、rabbitmq的介绍
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引入依赖。
<dependency> <groupid>org.springframework.boot</groupid> <artifactid>spring-boot-starter-amqp</artifactid> </dependency>
在application.properties中配置rabbitmq相关的信息,并首先启动了rabbitmq实例,并创建两个queue。
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的时候指定一个键,使其和交换机绑定。
@configuration public class rabbitconfig { @bean public org.springframework.amqp.core.queue queue() { return new org.springframework.amqp.core.queue("hello"); } }
接着就可以发送消息啦。在springboot中,我们使用amqptemplate去发送消息。代码如下:
@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); } }
生产者发送消息之后就需要消费者接收消息。这里定义了两个消息消费者,用来模拟生产者与消费者一对多的关系。
@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条消息。
@autowired private hellosender hellosender; @test public void hello() throws exception { for(int i=0;i<10;i++) { hellosender.send(i); } }
实际上rabbitmq还可以支持发送对象,当然由于涉及到序列化和反序列化,该对象要实现serilizable接口。这里定义了user对象,用来做发送消息内容。
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对象。
@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对象。
@component @rabbitlistener(queues = "object") public class modelrecevicer { @rabbithandler public void process(user user) { system.out.println("receiver object : " + user); } }
在单元测试中注入modelsender 对象,实例化user对象,然后发送。
@autowired private modelsender modelsender; @test public void model() throws exception { user user=new user("abc","123"); modelsender.sendmodel(user); }
三 、springboot整合rabbitmq(topic转发模式)
首先需要在rabbitmq服务端创建交换机topicexchange,并绑定两个queue:topic.message、topic.messages。
新建topicrabbitconfig,设置对应的queue与binding。
@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个消息。
@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); } }
生产者发送消息,这里创建了两个接收消息的消费者。
@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 发送消息。
@autowired private topicsender topicsender; @test public void topicsender() throws exception { topicsender.send(); topicsender.send1(); topicsender.send2(); }
从上面的输出结果可以看到,topic receiver2 匹配到了所有消息,topic receiver1只匹配到了1个消息。
四 、springboot整合rabbitmq(fanout exchange形式)
fanout exchange形式又叫广播形式,因此我们发送到路由器的消息会使得绑定到该路由器的每一个queue接收到消息。首先需要在rabbitmq服务端创建交换机fanoutexchange,并绑定三个queue:fanout.a、fanout.b、fanout.c。
与topic类似,新建fanoutrabbitconfig,绑定交换机和队列。
@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中发送消息。
@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。
@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); } }
在单元测试中注入消息发送者,发送消息。
@autowired private fanoutsender fanoutsender; @test public void fanoutsender() throws exception { fanoutsender.send(); }
从下图可以看到3个队列都接收到了消息。
本章节创建的类比较多,下图为本章节的结构,也可以直接查看demo源码了解。
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持。