消息发送、接收简单代码示例
mq.xml
//rabbitmq config
spring.rabbitmq.host=ip:port
spring.rabbitmq.username=
spring.rabbitmq.password=
spring.rabbitmq.virtual-host=
//发送队列
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;
@Bean
public ConnectionFactory connectionFactory() {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
connectionFactory.setAddresses(address);
connectionFactory.setUsername(username);
connectionFactory.setPassword(password);
connectionFactory.setVirtualHost(virtualHost);
connectionFactory.setPublisherConfirms(true); //必须要设置、消息发送确认
return connectionFactory;
}
/**
* 常用spring为singleton单例模式，此处mq消息需将其改为非单例模式
*/
@Bean
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)//必须是prototype类型
public RabbitTemplate rabbitTemplate() {
return new RabbitTemplate(connectionFactory());
}
@Bean
public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
return new RabbitAdmin(connectionFactory);
}
}
//消息发送
@Component
public class SystemMqMessageSender {
private static final Logger logger = LoggerFactory.getLogger(SystemMqMessageSender.class);
@Autowired
private AmqpTemplate rabbitTemplate;
@Value("${send.exchange.name}")
private String exchangeSystem;
@Value("${send.queue.name}")
private String queueSystem;
@Resource
private RabbitAdmin rabbitAdmin;
public void sendMessage(EventModel eventModel) {
String message = JsonUtils.json(eventModel);
logger.info("发送消息：{}", message);
rabbitTemplate.convertAndSend(exchangeSystem, queueSystem, message);
}
//声明持久化队列，并绑定到exchange上
@Bean
public Binding bindingExchangeSystem() {
Queue queue = QueueBuilder.durable(queueSystem).build();//队列持久化
rabbitAdmin.declareQueue(queue);//声明队列
DirectExchange exchange = (DirectExchange) ExchangeBuilder.directExchange(exchangeSystem).build();
rabbitAdmin.declareExchange(exchange);//创建路由
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;
@RabbitHandler
public void process(String message) {//监听消息
logger.info("接收到消息：{}", message);
processMessage(message, queueName);
}
public void processMessage(String content, String queueName) {
//业务处理
}
}rabbitmq如何保证高可用呢
答案是消息应答机制，一下是rabbitmq消息应答机制的原文：
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.
执行一个任务可能需要花费几秒钟，你可能会担心如果一个消费者在执行任务过程中挂掉了。一旦RabbitMQ将消息分发给了消费者，就会从内存中删除。在这种情况下，如果正在执行任务的消费者宕机，会丢失正在处理的消息和分发给这个消费者但尚未处理的消息。
但是，我们不想丢失任何任务，如果有一个消费者挂掉了，那么我们应该将分发给它的任务交付给另一个消费者去处理。
为了确保消息不会丢失，RabbitMQ支持消息应答。消费者发送一个消息应答，告诉RabbitMQ这个消息已经接收并且处理完毕了。RabbitMQ就可以删除它了。
如果一个消费者挂掉却没有发送应答，RabbitMQ会理解为这个消息没有处理完全，然后交给另一个消费者去重新处理。这样，你就可以确认即使消费者偶尔挂掉也不会丢失任何消息了。
没有任何消息超时限制；只有当消费者挂掉时，RabbitMQ才会重新投递。即使处理一条消息会花费很长的时间。
消息应答是默认打开的。我们通过显示的设置autoAsk=true关闭这种机制。现即自动应答开，一旦我们完成任务，消费者会自动发送应答。通知RabbitMQ消息已被处理，可以从内存删除。如果消费者因宕机或链接失败等原因没有发送ACK（不同于ActiveMQ，在RabbitMQ里，消息没有过期的概念），则RabbitMQ会将消息重新发送给其他监听在队列的下一个消费者。