1 Fanout
1.1 新建一个fanout类型的exchange
这里的创建都是用代码来创建，也可以在后台界面操作。
@Bean
FanoutExchange fanoutExchange() {
return new FanoutExchange("zenghongFanoutExchange");
}
启动服务后可以从控制台看到zenghongFanoutExchange已经创建成功：
image.png
1.2 再创建一个队列
@Bean
public Queue zenghongFanoutQueue() {
return new Queue("zenghongFanoutQueue");
}
image.png
1.3 绑定这个exchange和queue
@Bean
Binding bindingExchangeC(Queue zenghongFanoutQueue, FanoutExchange zenghongFanoutExchange) {
return BindingBuilder.bind(zenghongFanoutQueue).to(zenghongFanoutExchange);
}
image.png
1.4 生产者和消费者代码：
public void sendFanout(){
String sendMsg = "my Fanout msg " + new Date();
System.out.println("Sender1 : " + sendMsg);
this.rabbitTemplate.convertAndSend("zenghongFanoutExchange",null,sendMsg);
}
这里第二个参数是routing key ，我们传null。
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
/**
* @author: zenghong
* @Date: 2019/4/25 09:46
* @Description: TODO
*/
@Component
@RabbitListener(queues = "zenghongFanoutQueue")
public class FanoutCosumer {
@RabbitHandler
public void process(String msg) {
System.out.println("FanoutCosumer Receiver1 : " + msg);
}
}
1.5 测试接口
@RestController
@RequestMapping("/rabbit")
public class RabbitMqController {
@Autowired
private HelloSender helloSender1;
@PostMapping("/hello")
public String hello() {
helloSender1.send();
return "success";
}
@PostMapping("/myFanout")
public String myFanout() {
helloSender1.sendFanout();
return "myFanout success";
}
1.6 发送请求
image.png
image.png
消息被成功发送和接收，fanout果然是不需要routing key的。
2 direct
2.1 创建一个directExchange（zenghongDirectExchange） 和 两个 queue(zenghongDirectQueue1,zenghongDirectQueue2)，然后绑定zenghongDirectExchange 和 zenghongDirectQueue1，绑定zenghongDirectExchange 和 zenghongDirectQueue2。
@Bean
DirectExchange zenghongDirectExchange() {
return new DirectExchange("zenghongDirectExchange");
}
@Bean
public Queue zenghongDirectQueue1() {
return new Queue("zenghongDirectQueue1");
}
@Bean
public Queue zenghongDirectQueue2() {
return new Queue("zenghongDirectQueue2");
}
@Bean
Binding bindingExchange2(Queue zenghongDirectQueue1, DirectExchange zenghongDirectExchange) {
return BindingBuilder.bind(zenghongDirectQueue1).to(zenghongDirectExchange).with("direct queue1");
}
@Bean
Binding bindingExchange3(Queue zenghongDirectQueue2, DirectExchange zenghongDirectExchange) {
return BindingBuilder.bind(zenghongDirectQueue2).to(zenghongDirectExchange).with("direct queue2");
}
2.2 编写sender、receiver和测试接口
sender：
public void sendDirect(){
String sendMsg = "my Direct msg " + new Date();
System.out.println("Sender : " + sendMsg);
this.rabbitTemplate.convertAndSend("zenghongDirectExchange","direct queue1",sendMsg);
}
注意：这里send发送的routing key和 queue1的binding key完全比配。
receiver:
@Component
@RabbitListener(queues = "zenghongDirectQueue1")
public class DirectCosumer1 {
@RabbitHandler
public void process(String msg) {
System.out.println("DirectCosumer1 Receiver1 : " + msg);
}
}
@Component
@RabbitListener(queues = "zenghongDirectQueue2")
public class DirectCosumer2 {
@RabbitHandler
public void process(String msg) {
System.out.println("DirectCosumer2 Receiver1 : " + msg);
}
}
接口：
@PostMapping("/myDirect")
public String myDirect() {
helloSender1.sendDirect();
return "myDirect success";
}
2.3 测试验证：
image.png
控制台：
image.png
和我们预想的一样，监听queue1的DirectCosumer1接收到了消息。
3 topic
3.1 创建exchange、queue并绑定：
@Bean
TopicExchange zenghongTopicExchange() {
return new TopicExchange("zenghongTopicExchange");
}
@Bean
public Queue zenghongTopicQueue1() {
return new Queue("zenghongTopicQueue1");
}
@Bean
public Queue zenghongTopicQueue2() {
return new Queue("zenghongTopicQueue2");
}
@Bean
Binding bindingExchange2(Queue zenghongTopicQueue1, TopicExchange zenghongTopicExchange) {
return BindingBuilder.bind(zenghongTopicQueue1).to(zenghongTopicExchange).with("queue1.#");
}
@Bean
Binding bindingExchange3(Queue zenghongTopicQueue2, TopicExchange zenghongTopicExchange) {
return BindingBuilder.bind(zenghongTopicQueue2).to(zenghongTopicExchange).with("queue1.*");
}
3.2 sender 和 cosumer
public void sendTopic(){
String sendMsg = "my Topic msg " + new Date();
System.out.println("Sender : " + sendMsg);
this.rabbitTemplate.convertAndSend("zenghongTopicExchange","queue1.hahahaha",sendMsg);
}
3.3 测试代码和测试
@PostMapping("/myTopic")
public String myTopic() {
helloSender1.sendTopic();
return "myTopic success";
}
image.png
image.png
和我们预想的一样，queue1.hahahaha（routing key ） 和 queue1.# 、 queue1.* 都匹配上了，所以两个队列的cosumer都收到了消息。
4 header
4.1 创建queue exchange 并绑定
@Bean
HeadersExchange zenghongHeadersExchange() {
return new HeadersExchange("zenghongHeadersExchange");
}
@Bean
public Queue zenghongHeadersQueue() {
return new Queue("zenghongHeadersQueue");
}
@Bean
Binding bindingExchange6(Queue zenghongHeadersQueue, HeadersExchange zenghongHeadersExchange) {
Map<String,Object> map=new HashMap<String,Object>();
map.put("test1","test");
return BindingBuilder.bind(zenghongHeadersQueue).to(zenghongHeadersExchange).whereAll(map).match();
}
4.2 sender和cosumer
public void sendHeaders(){
String sendMsg = "my Headers msg " + new Date();
System.out.println("Sender : " + sendMsg);
Message msg= null;
try {
msg = MessageBuilder.withBody(sendMsg.getBytes("utf-8")).
setHeader("test1","test").build();
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
this.rabbitTemplate.convertAndSend("zenghongHeadersExchange",null,
msg);
}
@Component
@RabbitListener(queues = "zenghongHeadersQueue")
public class HeadersCosumer {
@RabbitHandler
public void process(Message message) {
try {
System.out.println("HeadersCosumer Receiver : " + new String(message.getBody(),"utf-8"));
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
}
}
4.3 测试代码并测试
@PostMapping("/myHeaders")
public String myHeaders() {
helloSender1.sendHeaders();
return "myHeaders success";
}
image.png
然后我们发现报错了，而且在刷屏！
org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException: Listener method 'no match' threw exception
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:198) ~[spring-rabbit-2.1.5.RELEASE.jar:2.1.5.RELEASE]
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.onMessage(MessagingMessageListenerAdapter.java:127) ~[spring-rabbit-2.1.5.RELEASE.jar:2.1.5.RELEASE]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1552) ~[spring-rabbit-2.1.5.RELEASE.jar:2.1.5.RELEASE]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1478) ~[spring-rabbit-2.1.5.RELEASE.jar:2.1.5.RELEASE]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1466) ~[spring-rabbit-2.1.5.RELEASE.jar:2.1.5.RELEASE]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1461) ~[spring-rabbit-2.1.5.RELEASE.jar:2.1.5.RELEASE]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1410) ~[spring-rabbit-2.1.5.RELEASE.jar:2.1.5.RELEASE]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:870) [spring-rabbit-2.1.5.RELEASE.jar:2.1.5.RELEASE]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:854) [spring-rabbit-2.1.5.RELEASE.jar:2.1.5.RELEASE]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:78) [spring-rabbit-2.1.5.RELEASE.jar:2.1.5.RELEASE]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1137) [spring-rabbit-2.1.5.RELEASE.jar:2.1.5.RELEASE]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1043) [spring-rabbit-2.1.5.RELEASE.jar:2.1.5.RELEASE]
at java.lang.Thread.run(Thread.java:748) [na:1.8.0_131]
Caused by: org.springframework.amqp.AmqpException: No method found for class [B
at org.springframework.amqp.rabbit.listener.adapter.DelegatingInvocableHandler.getHandlerForPayload(DelegatingInvocableHandler.java:149) ~[spring-rabbit-2.1.5.RELEASE.jar:2.1.5.RELEASE]
at org.springframework.amqp.rabbit.listener.adapter.DelegatingInvocableHandler.invoke(DelegatingInvocableHandler.java:129) ~[spring-rabbit-2.1.5.RELEASE.jar:2.1.5.RELEASE]
at org.springframework.amqp.rabbit.listener.adapter.HandlerAdapter.invoke(HandlerAdapter.java:60) ~[spring-rabbit-2.1.5.RELEASE.jar:2.1.5.RELEASE]
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:190) ~[spring-rabbit-2.1.5.RELEASE.jar:2.1.5.RELEASE]
... 12 common frames omitted
为啥在刷屏呢？看看官方说的
If retries are not enabled and the listener throws an exception, by default the delivery will be retried indefinitely. You can modify this behavior in two ways; set the defaultRequeueRejected
property to false
and zero re-deliveries will be attempted; or, throw an AmqpRejectAndDontRequeueException
to signal the message should be rejected. This is the mechanism used when retries are enabled and the maximum delivery attempts are reached.
还好我们有百度翻译：
如果未启用重试，并且侦听器引发异常，则默认情况下将无限期重试传递。您可以通过两种方式修改此行为；设置默认的Requerejected
属性改为假
并且将尝试零次重新传递；或者，引发amqpReject和donTrequeueException
要发出信号，应拒绝该消息。这是启用重试并达到最大传递尝试次数时使用的机制。
知道为啥要刷屏了，为啥要报错呢？ 查了一下，需要消费者把@RabbitListener放到方法上。
@Component
public class HeadersCosumer {
@RabbitListener(queues = "zenghongHeadersQueue")
@RabbitHandler
public void process(Message message) {
try {
System.out.println("HeadersCosumer Receiver : " + new String(message.getBody(),"utf-8"));
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
}
}
image.png
成功接收到了消息。
