网站如何建设二级域名代理,游戏推广员怎么做,微信小商店分销系统,想成为网站设计师要怎么做前言
前面学习了 RabbitMQ 基础#xff0c;现在主要记录下学习 Spring Boot 整合 RabbitMQ #xff0c;调用它的 API #xff0c;以及中间使用的相关功能的记录。
相关的可以去[我的博客/RabbitMQ]
正文
我这里测试都是使用的是 topic 交换器#xff0c;Spring Boot 2.…前言
前面学习了 RabbitMQ 基础现在主要记录下学习 Spring Boot 整合 RabbitMQ 调用它的 API 以及中间使用的相关功能的记录。
相关的可以去[我的博客/RabbitMQ]
正文
我这里测试都是使用的是 topic 交换器Spring Boot 2.0.0 jdk 1.8
配置
Spring Boot 版本 2.0.0 在 pom.xml 文件中引入 AMQP 的依赖
dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId
/dependency在系统配置文件中加入连接属性
spring:application:name: RabbitMQ-Demorabbitmq:host: k.wuwii.comport: 5672username: kronchanpassword: 123456#virtual-host: testpublisher-confirms: true # 开启确认消息是否到达交换器需要设置 truepublisher-returns: true # 开启确认消息是否到达队列需要设置 true基本的使用
消费者
新增一个消费者类
Log
public class MessageReceiver implements ChannelAwareMessageListener {Overridepublic void onMessage(Message message, Channel channel) throws Exception {try {byte[] body message.getBody();log.info( receive new String(body));} finally {// 确认成功消费否则消息会转发给其他的消费者或者进行重试channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);}}
}配置类
新增 RabbitMQ 的配置类主要是对消费者的队列交换器路由键的一些设置
Configuration
public class RabbitMQConfig {public final static String QUEUE_NAME springboot.demo.test1;public final static String ROUTING_KEY route-key;public final static String EXCHANGES_NAME demo-exchanges;Beanpublic Queue queue() {// 是否持久化boolean durable true;// 仅创建者可以使用的私有队列断开后自动删除boolean exclusive false;// 当所有消费客户端连接断开后是否自动删除队列boolean autoDelete false;return new Queue(QUEUE_NAME, durable, exclusive, autoDelete);}/*** 设置交换器这里我使用的是 topic exchange*/Beanpublic TopicExchange exchange() {// 是否持久化boolean durable true;// 当所有消费客户端连接断开后是否自动删除队列boolean autoDelete false;return new TopicExchange(EXCHANGES_NAME, durable, autoDelete);}/*** 绑定路由*/Beanpublic Binding binding(Queue queue, TopicExchange exchange) {return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY);}Beanpublic SimpleMessageListenerContainer container(ConnectionFactory connectionFactory) {SimpleMessageListenerContainer container new SimpleMessageListenerContainer();container.setConnectionFactory(connectionFactory);container.setQueueNames(QUEUE_NAME);container.setMessageListener(receiver());//container.setMaxConcurrentConsumers(1);//container.setConcurrentConsumers(1); 默认为1//container.setExposeListenerChannel(true);container.setAcknowledgeMode(AcknowledgeMode.MANUAL); // 设置为手动默认为 AUTO如果设置了手动应答 basicack就要设置manualreturn container;}Beanpublic MessageReceiver receiver() {return new MessageReceiver();}}生产者
Component
public class MessageSender {Autowiredprivate RabbitTemplate rabbitTemplate;/*** logger*/private static final Logger log LoggerFactory.getLogger(MessageSender.class);public void send() {// public void convertAndSend(String exchange, String routingKey, final Object object, CorrelationData correlationData)// exchange: 交换机名称// routingKey: 路由关键字// object: 发送的消息内容// correlationData:消息IDCorrelationData correlationId new CorrelationData(UUID.randomUUID().toString());// ConfirmListener是当消息无法发送到Exchange被触发此时Ack为False这时cause包含发送失败的原因例如exchange不存在时// 需要在系统配置文件中设置 publisher-confirms: trueif (!rabbitTemplate.isConfirmListener()) {rabbitTemplate.setConfirmCallback((correlationData, ack, cause) - {if (ack) {log.info( 消息id:{} 发送成功, correlationData.getId());} else {log.info( 消息id:{} 发送失败, correlationData.getId());}});}// ReturnCallback 是在交换器无法将路由键路由到任何一个队列中会触发这个方法。// 需要在系统配置文件中设置 publisher-returns: truerabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) - {log.info(消息id{} 发送失败, message.getMessageProperties().getCorrelationId());});rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGES_NAME, RabbitMQConfig.ROUTING_KEY, Hello World, correlationId);log.info(Already sent message.);}}测试发送消息
先启动系统启动类消费者开始订阅启动测试类发送消息。
RunWith(SpringRunner.class)
SpringBootTest
public class SpringbootRabbitmqApplicationTests {Autowiredprivate MessageSender sender;Testpublic void testReceiver() {sender.send();}
}可以在消费者接收到信息并且发送端将打出日志 成功发送消息的记录也可以测试下 Publisher Confirms and Returns机制 主要是测试 ConfirmCallback 和 ReturnCallback 这两个方法。
ConfirmCallback 确认消息是否到达交换器例如我们发送一个消息到一个你没有创建过的 交换器上面去看看情况ReturnCallback确认消息是否到达队列我们可以这样测试定义一个路由键不会被任何队列订阅到最后查看结果就可以了。
使用注解的方式
引入依赖和连接参数
跟文章第一步的配置一样的。
消费者
Component
Log
public class MessageReceiver {/*** 无返回消息的** param message*/RabbitListener(bindings QueueBinding(value Queue(value Constant.QUEUE_NAME, durable true, exclusive false, autoDelete false),exchange Exchange(value Constant.EXCHANGES_NAME, ignoreDeclarationExceptions true, type ExchangeTypes.TOPIC, autoDelete false),key Constant.ROUTING_KEY))public void receive(byte[] message) {log.info( receive new String(message));}/*** 设置有返回消息的* 需要注意的是* 1. 在消息的在生产者发送消息端一定要使用 SendAndReceive(……) 这种带有 receive 的方法否则会抛异常不捕获会死循环。* 2. 该方法调用时会锁定当前线程并且有可能会造成MQ的性能下降或者服务端/客户端出现死循环现象请谨慎使用。** param message* return*/RabbitListener(bindings QueueBinding(value Queue(value Constant.QUEUE_NAME, durable true, exclusive false, autoDelete false),exchange Exchange(value Constant.EXCHANGES_NAME, ignoreDeclarationExceptions true, type ExchangeTypes.TOPIC, autoDelete false),key Constant.ROUTING_REPLY_KEY))public String receiveAndReply(byte[] message) {log.info( receive new String(message));return I got the message;}}主要是使用到 RabbitListener虽然看起来参数很多仔细的你会发现这个和写配置类里面的基本属性是一摸一样的没有任何区别。
需要注意的是我在这里多做了个有返回值的消息这个使用异常的话会不断重试消息从而阻塞了线程。而且使用它的时候只能使用带有 receive 的方法给它发送消息。
生产者
生产者没什么变化。
Component
public class MessageSender implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {/*** logger*/private static final Logger log LoggerFactory.getLogger(MessageSender.class);private RabbitTemplate rabbitTemplate;/*** 注入 RabbitTemplate*/Autowiredpublic MessageSender(RabbitTemplate rabbitTemplate) {this.rabbitTemplate rabbitTemplate;this.rabbitTemplate.setConfirmCallback(this);this.rabbitTemplate.setReturnCallback(this);}/*** 测试无返回消息的*/public void send() {CorrelationData correlationData new CorrelationData(UUID.randomUUID().toString());rabbitTemplate.convertAndSend(Constant.EXCHANGES_NAME, Constant.ROUTING_KEY, Hello World.getBytes(), correlationData);log.info( Already sent message);}/*** 测试有返回消息的需要注意一些问题*/public void sendAndReceive() {CorrelationData correlationData new CorrelationData(UUID.randomUUID().toString());Object o rabbitTemplate.convertSendAndReceive(Constant.EXCHANGES_NAME, Constant.ROUTING_REPLY_KEY, Hello World Second.getBytes(), correlationData);log.info( {}, Objects.toString(o));}/*** Confirmation callback.** param correlationData correlation data for the callback.* param ack true for ack, false for nack* param cause An optional cause, for nack, when available, otherwise null.*/Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {if (ack) {log.info( 消息id:{} 发送成功, correlationData.getId());} else {log.info( 消息id:{} 发送失败, correlationData.getId());}}/*** Returned message callback.** param message the returned message.* param replyCode the reply code.* param replyText the reply text.* param exchange the exchange.* param routingKey the routing key.*/Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {log.info(消息id{} 发送失败, message.getMessageProperties().getCorrelationId());}
}测试
RunWith(SpringRunner.class)
SpringBootTest
public class SpringbootAnnotationApplicationTests {Autowiredprivate MessageSender sender;Testpublic void send() {sender.send();}Testpublic void sendAndReceive() {sender.sendAndReceive();}
}
