------------------------------RabbitMQ------------------------
RabbitMQ队列服务由四部分组成：发送消息者，交换机，队列和接受消息者。
发送消息者负责生产消息并将消息发送给指定的交换机；
交换机根据一定的调度策略把消息丢到绑定的队列上去或者直接丢弃，它不会存储消息；
队列负责同步的传送消息；
接受消息者从队列获取到消息，并做进一步处理；
交换机根据调度策略的差异分为四种类型:
Direct：先匹配再投递，只有消息发送者指定的队列key和队列绑定时指定的key相同时，才会投递到该队列；
Topic：按照一定的规则投递，使用较灵活
Headers：也是按照一定的规则匹配的交换机
Fanout：投递消息到所有绑定队列
1.安装RabbitMQ
brew install rabbitmq
安装完的目录在/usr/local/Cellar/rabbitmq.
2.启动RabbitMQ和插件
屏幕快照 2017-06-20 下午3.19.13.png
执行一次即可，以后都不需要执行
3.管理界面
http://localhost:15672/
默认的用户名和密码都是guest
----------------------------------------SpringBoot结合--------------------
新建一个SpringBoot项目，勾选AMQP服务即可。或者手动添加spring-boot-starter-amqp依赖；
application.properties文件：
spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
除此以外，无需其他配置。
1. Direct
1.配置队列
package com.example.demo;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitConfig {
@Bean
public Queue Queue() {
return new Queue("Test"); //队列名称
}
}
2.消息产生者/发送者
package com.example.demo;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;
import java.util.Date;
@Component
public class Sender {
@Autowired
private AmqpTemplate amqpTemplate;
public void send() {
String content = "hello world";
System.out.println("Sender: " + content);
this.amqpTemplate.convertAndSend("Test", content); //生产者和消费者的队列名称必须保持一致，否则不能接受到消息
}
}
3.消息接受者
package com.example.demo;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
@RabbitListener(queues = "Test") //生产者和消费者的队列名称必须保持一致，否则不能接受到消息
public class Receiver {
@RabbitHandler
public void process(String txt) {
System.out.println("Reciver:" + txt);
}
}
4.测试
package com.example.demo;
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;
@RunWith(SpringRunner.class)
@SpringBootTest
public class DemoApplicationTests {
@Autowired
private Sender sender;
@Test
public void hello() throws Exception {
sender.send();
}
}
上面的是简单的一对一发送消息。
一对多发送
再添加一个消息接受者Receiver2，
将测试部分改为：
for (int i=0; i<100; i++) {
sender.send(i);
}
一个发送者，两个接收者，消息会均匀的发送到两个接收者。
多对多发送
再添加一个消息发送者Sender2,
测试部分为：
@Autowired
private Sender sender;
@Autowired
private Sender2 sender2;
@Test
public void hello() throws Exception {
for (int i=0; i<100; i++) {
sender.send(i);
sender2.send(i);
}
}
消息接收者仍会均匀的接收到消息。
传递自定义对象
自定义一个User类：
package com.example.demo;
import org.springframework.stereotype.Component;
import java.io.Serializable;
import java.security.Principal;
@Component
public class User implements Serializable {
private String name;
private int age;
public User() {
super();
}
public User(String name, int age) {
this.name = name;
this.age = age;
}
@Override
public String toString() {
return "User{" +
"name='" + name + '\'' +
", age=" + age +
'}';
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public int getAge() {
return age;
}
public void setAge(int age) {
this.age = age;
}
}
发送者：
public void sender(User user) {
System.out.println("Sender : " + user);
this.amqpTemplate.convertAndSend("hello", user);
}
接收者：
@RabbitHandler
public void process(User user) {
System.out.println("Reciver : " + user);
}
注意：传递自定义对象消息，比如User, User类一定要实现Serializable接口，否则会报如下错误：
[cTaskExecutor-1] s.a.r.l.ConditionalRejectingErrorHandler : Execution of Rabbit message listener failed.
Caused by: org.springframework.amqp.AmqpException: No method found for class [B
并可能导致死循环。
2.Topic Exchange方式
Topic Exchange方式是比较灵活的一种。它转发消息主要靠通配符，只有通配符匹配之后才会转发。
这时候的路由键必须是用.点分的一串字符，通配符中*表示一个词. #表示零个或多个词.
比如test.a.*只能匹配第一个词是test，第二个是a的三个词的路由键；
test.a.#则可以匹配任意由test.a开头的路由键。
首先，配置路由键及topic规则：
package com.example.demo;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@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("exchange");
}
@Bean
Binding bingdingExchangeMessage(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.#");
}
}
发送者：
package com.example.demo;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Component
public class Sender {
@Autowired
private AmqpTemplate amqpTemplate;
public void send1() {
String content = "message 1";
System.out.println("sender send message 1");
this.amqpTemplate.convertAndSend("exchange", "topic.message", content); //会匹配到topic.#和topic.message 两个Receiver都可以收到消息
}
public void send2() {
String content = "message 2";
System.out.println("sender send message 2");
this.amqpTemplate.convertAndSend("exchange", "topic.messages", content);//只有topic.#可以匹配到，所以只有Receiver2监听到消息
}
}
接收者（2个）：
package com.example.demo;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
@RabbitListener(queues = TopicRabbitConfig.message)
public class Reciver1 {
@RabbitHandler
public void process(String txt) {
System.out.println("reciver 1 : " + txt);
}
}
package com.example.demo;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
@RabbitListener(queues = TopicRabbitConfig.messages)
public class Reciver2 {
@RabbitHandler
public void process(String txt) {
System.out.println("reciver 2: " + txt);
}
}
测试：
package com.example.demo;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
@SpringBootApplication
public class DemoApplication {
@Autowired
Sender sender;
@RequestMapping(value = "/go")
public void sender() {
sender.send1();
sender.send2();
}
public static void main(String[] args) {
SpringApplication.run(DemoApplication.class, args);
}
}
3.Fanout Exchange
Fanout Exchange采用的是一种广播策略，会把消息转发到所有绑定的队列上去。
路由键指定及交换机绑定(指定三个队列)
package com.example.demo;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class FanoutExchangeConf {
@Bean
public Queue AMessage() {
return new Queue("faout.A");
}
@Bean
public Queue BMessage() {
return new Queue("faout.B");
}
@Bean
Queue CMessage() {
return new Queue("faout.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);
}
}
发送者：
package com.example.demo;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;
@Component
public class Sender {
@Autowired
AmqpTemplate amqpTemplate;
public void send() {
String content = "message from sender";
System.out.println("Sender : " + content);
this.amqpTemplate.convertAndSend("fanoutExchange","",content);
}
}
三个接收者，分别制定三个队列faout.A faout.B faout.C
@Component
@RabbitListener(queues = "faout.A")
public class ReciverA {
@RabbitHandler
public void process(String txt) {
System.out.println("reciver A: " + txt);
}
}
测试：
@Autowired
Sender sender;
@RequestMapping(value = "/go")
public void sender() {
sender.send();
}
三个接收者都会接受到发送者发送的消息。
参考：
RabbitMQ 使用参考
RabbitMQ：Spring 集成 RabbitMQ 与其概念，消息持久化，ACK机制等
springboot(八)：RabbitMQ详解
最后帮朋友打个小广告
