消息發(fā)送、接收簡單代碼示例

mq.xml//rabbitmq config spring.rabbitmq.host=ip:port spring.rabbitmq.username= spring.rabbitmq.password= spring.rabbitmq.virtual-host=//發(fā)送隊列 send.exchange.name= send.queue.name=//接收listen.queue.name.system=@Configuration public class AmqpConfig {@Value("${spring.rabbitmq.host}")private String address;@Value("${spring.rabbitmq.username}")private String username;@Value("${spring.rabbitmq.password}")private String password;@Value("${spring.rabbitmq.virtual-host}")private String virtualHost;@Beanpublic ConnectionFactory connectionFactory() {CachingConnectionFactory connectionFactory = new CachingConnectionFactory();connectionFactory.setAddresses(address);connectionFactory.setUsername(username);connectionFactory.setPassword(password);connectionFactory.setVirtualHost(virtualHost);connectionFactory.setPublisherConfirms(true); //必須要設(shè)置、消息發(fā)送確認return connectionFactory;}/*** 常用spring為singleton單例模式，此處mq消息需將其改為非單例模式*/@Bean@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)//必須是prototype類型public RabbitTemplate rabbitTemplate() {return new RabbitTemplate(connectionFactory());}@Beanpublic RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {return new RabbitAdmin(connectionFactory);}}//消息發(fā)送 @Component public class SystemMqMessageSender {private static final Logger logger = LoggerFactory.getLogger(SystemMqMessageSender.class);@Autowiredprivate AmqpTemplate rabbitTemplate;@Value("${send.exchange.name}")private String exchangeSystem;@Value("${send.queue.name}")private String queueSystem;@Resourceprivate RabbitAdmin rabbitAdmin;public void sendMessage(EventModel eventModel) {String message = JsonUtils.json(eventModel);logger.info("發(fā)送消息：{}", message);rabbitTemplate.convertAndSend(exchangeSystem, queueSystem, message);}//聲明持久化隊列，并綁定到exchange上@Beanpublic Binding bindingExchangeSystem() {Queue queue = QueueBuilder.durable(queueSystem).build();//隊列持久化rabbitAdmin.declareQueue(queue);//聲明隊列DirectExchange exchange = (DirectExchange) ExchangeBuilder.directExchange(exchangeSystem).build();rabbitAdmin.declareExchange(exchange);//創(chuàng)建路由Binding binding = BindingBuilder.bind(queue).to(exchange).withQueueName();//綁定路由rabbitAdmin.declareBinding(binding);return binding;}}//消息接收 @Component @RabbitListener(queues = "${listen.queue.name.system}") public class SystemMessageListener extends BaseListener implements EventModelConsumer,InitializingBean {private static final Logger logger = LoggerFactory.getLogger(SystemMessageListener.class);@Value("${listen.queue.name.system}")private String queueName;@RabbitHandlerpublic void process(String message) {//監(jiān)聽消息logger.info("接收到消息：{}", message);processMessage(message, queueName);}public void processMessage(String content, String queueName) {//業(yè)務(wù)處理} }

rabbitmq如何保證高可用呢

答案是消息應(yīng)答機制，一下是rabbitmq消息應(yīng)答機制的原文：
Doing a task can take a few seconds. You may wonder what happens if one of the consumers starts a long task and dies with it only partly done. With our current code, once RabbitMQ delivers a message to the customer it immediately marks it for deletion. In this case, if you kill a worker we will lose the message it was just processing. We'll also lose all the messages that were dispatched to this particular worker but were not yet handled.

But we don't want to lose any tasks. If a worker dies, we'd like the task to be delivered to another worker.

In order to make sure a message is never lost, RabbitMQ supports message acknowledgments. An ack(nowledgement) is sent back by the consumer to tell RabbitMQ that a particular message has been received, processed and that RabbitMQ is free to delete it.

If a consumer dies (its channel is closed, connection is closed, or TCP connection is lost) without sending an ack, RabbitMQ will understand that a message wasn't processed fully and will re-queue it. If there are other consumers online at the same time, it will then quickly redeliver it to another consumer. That way you can be sure that no message is lost, even if the workers occasionally die.

There aren't any message timeouts; RabbitMQ will redeliver the message when the consumer dies. It's fine even if processing a message takes a very, very long time.

Manual message acknowledgments are turned on by default. In previous examples we explicitly turned them off via the autoAck=true flag. It's time to set this flag to false and send a proper acknowledgment from the worker, once we're done with a task.
執(zhí)行一個任務(wù)可能需要花費幾秒鐘，你可能會擔心如果一個消費者在執(zhí)行任務(wù)過程中掛掉了。一旦RabbitMQ將消息分發(fā)給了消費者，就會從內(nèi)存中刪除。在這種情況下，如果正在執(zhí)行任務(wù)的消費者宕機，會丟失正在處理的消息和分發(fā)給這個消費者但尚未處理的消息。
但是，我們不想丟失任何任務(wù)，如果有一個消費者掛掉了，那么我們應(yīng)該將分發(fā)給它的任務(wù)交付給另一個消費者去處理。

為了確保消息不會丟失，RabbitMQ支持消息應(yīng)答。消費者發(fā)送一個消息應(yīng)答，告訴RabbitMQ這個消息已經(jīng)接收并且處理完畢了。RabbitMQ就可以刪除它了。

如果一個消費者掛掉卻沒有發(fā)送應(yīng)答，RabbitMQ會理解為這個消息沒有處理完全，然后交給另一個消費者去重新處理。這樣，你就可以確認即使消費者偶爾掛掉也不會丟失任何消息了。

沒有任何消息超時限制；只有當消費者掛掉時，RabbitMQ才會重新投遞。即使處理一條消息會花費很長的時間。

消息應(yīng)答是默認打開的。我們通過顯示的設(shè)置autoAsk=true關(guān)閉這種機制。現(xiàn)即自動應(yīng)答開，一旦我們完成任務(wù)，消費者會自動發(fā)送應(yīng)答。通知RabbitMQ消息已被處理，可以從內(nèi)存刪除。如果消費者因宕機或鏈接失敗等原因沒有發(fā)送ACK（不同于ActiveMQ，在RabbitMQ里，消息沒有過期的概念），則RabbitMQ會將消息重新發(fā)送給其他監(jiān)聽在隊列的下一個消費者。

轉(zhuǎn)載于:https://www.cnblogs.com/canmeng-cn/p/8543113.html

總結(jié)

以上是生活随笔為你收集整理的rabbitmq简单收发服务搭建的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。