自动配置信息 这里我开启ACK消息确认
server.port=8083
#服务器配置
spring.application.name=rabbitmq-hello-sending
#rabbitmq连接参数
spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=linpeng
spring.rabbitmq.password=123456
# 开启发送确认
spring.rabbitmq.publisher-confirms=true
# 开启发送失败退回
spring.rabbitmq.publisher-returns=true
# 开启ACK
spring.rabbitmq.listener.direct.acknowledge-mode=manual
spring.rabbitmq.listener.simple.acknowledge-mode=manual
创建消息队列 队列名：hello 和 helloObj
package com.example.demo;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitConfig {
@Bean
public Queue QueueA() {
return new Queue("hello");
}
@Bean
public Queue QueueB() {
return new Queue("helloObj");
}
/**
* Fanout 就是我们熟悉的广播模式或者订阅模式，给Fanout交换机发送消息，绑定了这个交换机的所有队列都收到这个消息。
* @return
*/
@Bean
FanoutExchange fanoutExchange() {
return new FanoutExchange("ABExchange");
}
@Bean
Binding bindingExchangeA(Queue QueueA, FanoutExchange fanoutExchange) {
return BindingBuilder.bind(QueueA).to(fanoutExchange);
}
@Bean
Binding bindingExchangeB(Queue QueueB, FanoutExchange fanoutExchange) {
return BindingBuilder.bind(QueueB).to(fanoutExchange);
}
}
消息发送者 Sender 使用 RabbitTemplate 不采用 AmqpTemplate
package com.example.demo;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;
import java.util.Date;
//RabbitTemplate.ConfirmCallback
@Service
public class HelloSender implements RabbitTemplate.ReturnCallback {
@Autowired
// private AmqpTemplate rabbitTemplate;
private RabbitTemplate rabbitTemplate;
public void send() {
String context = "你好现在是 " + new Date() +"";
System.out.println("HelloSender发送内容 : " + context);
// this.rabbitTemplate.setConfirmCallback(this);
this.rabbitTemplate.setReturnCallback(this);
this.rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
if (!ack) {
System.out.println("HelloSender消息发送失败" + cause + correlationData.toString());
} else {
System.out.println("HelloSender 消息发送成功 ");
}
});
this.rabbitTemplate.convertAndSend("hello", context);
}
public void sendObj() {
MessageObj obj = new MessageObj();
obj.setACK(false);
obj.setId(123);
obj.setName("zhangsan");
obj.setValue("data");
System.out.println("发送 : " + obj);
this.rabbitTemplate.convertAndSend("helloObj", obj);
}
@Override
public void returnedMessage(Message message, int i, String s, String s1, String s2) {
System.out.println("sender return success" + message.toString()+"==="+i+"==="+s1+"==="+s2);
}
// @Override
// public void confirm(CorrelationData correlationData, boolean b, String s) {
// System.out.println("sender success");
// }
}
消息接受者 Receiver 注解方式接受消息
package com.example.demo;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.annotation.RabbitListenerConfigurer;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistrar;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Configurable;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.util.Date;
import java.util.Map;
@Component
@RabbitListener(queues = "hello")
public class HelloReceiver {
@RabbitHandler
public void process(String hello,Channel channel, Message message) throws IOException {
System.out.println("HelloReceiver收到 : " + hello +"收到时间"+new Date());
try {
//告诉服务器收到这条消息 已经被我消费了 可以在队列删掉 这样以后就不会再发了 否则消息服务器以为这条消息没处理掉 后续还会在发
channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
System.out.println("receiver success");
} catch (IOException e) {
e.printStackTrace();
//丢弃这条消息
//channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,false);
System.out.println("receiver fail");
}
}
}
备注：我们用注解的方式来接受消息 就不要用 自己创建对象实现ChannelAwareMessageListener的方式来接受消息 这种方式还要去全局里面配置 麻烦，直接用@RabbitListener(queues = "hello")最简单
消息确认 因为我在属性配置文件里面开启了ACK确认 所以如果代码没有执行ACK确认 你在RabbitMQ的后台会看到消息会一直留在队列里面未消费掉 只要程序一启动开始接受该队列消息的时候 又会收到
channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
//消息的标识，false只确认当前一个消息收到，true确认所有consumer获得的消息
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
//ack返回false，并重新回到队列，api里面解释得很清楚
channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
//拒绝消息
channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
TestController测试
@Autowired
private HelloSender helloSender;
/**
* 单生产者-单个消费者
*/
@RequestMapping("/test")
public void hello() throws Exception {
helloSender.send();
}
RabbitMQ后台 两个队列
image.png
发送消息
image.png
ACK场景测试
我们把HelloReceiver的ACK确认代码注释掉 那消息就算程序收到了 但是未确认ACK导致消息服务器以为他是未成功消费的 后续还会再发
image.png
重启程序
image.png
网友评论