一 、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引入依赖。
|
1 2 3 4 |
<dependency> <groupid>org.springframework.boot</groupid> <artifactid>spring-boot-starter-amqp</artifactid> </dependency> |
在application.properties中配置rabbitmq相关的信息，并首先启动了rabbitmq实例，并创建两个queue。
|
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); } } |
实际上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(topic转发模式)
首先需要在rabbitmq服务端创建交换机topicexchange，并绑定两个queue：topic.message、topic.messages。
新建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(); } |
从上面的输出结果可以看到，topic receiver2 匹配到了所有消息，topic receiver1只匹配到了1个消息。
四 、springboot整合rabbitmq(fanout exchange形式)
fanout exchange形式又叫广播形式，因此我们发送到路由器的消息会使得绑定到该路由器的每一个queue接收到消息。首先需要在rabbitmq服务端创建交换机fanoutexchange，并绑定三个queue：fanout.a、fanout.b、fanout.c。
与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个队列都接收到了消息。
本章节创建的类比较多，下图为本章节的结构，也可以直接查看demo源码了解。
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。
原文链接：http://www.cnblogs.com/5ishare/p/10163318.html
查看更多关于SpringBoot之RabbitMQ的使用方法的详细内容...