Java搭建RabbitMq消息中间件过程详解
这篇文章主要介绍了Java搭建RabbitMq消息中间件过程详解,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
前言
当系统中出现“生产“和“消费“的速度或稳定性等因素不一致的时候，就需要消息队列。
名词
- exchange：交换机
- routingkey：路由key
- queue：队列
控制台端口：15672
exchange和queue是需要绑定在一起的，然后消息发送到exchange再由exchange通过routingkey发送到对应的队列中。
使用场景
1.技能订单3分钟自动取消，改变状态
2.直播开始前15分钟提醒
3.直播状态自动结束
流程
生产者发送消息—>order_pre_exchange交换机—>order_per_ttl_delay_queue队列
—>时间到期—>order_delay_exchange交换机—>order_delay_process_queue队列—>消费者
第一步：在pom文件中添加
org.springframework.boot spring-boot-starter-amqp
第二步：在application.properties文件中添加
spring.rabbitmq.host=172.xx.xx.xxx spring.rabbitmq.port=5672 spring.rabbitmq.username=rabbit spring.rabbitmq.password=123456 spring.rabbitmq.virtual-host=/ spring.rabbitmq.connection-timeout=15000 spring.rabbitmq.publisher-confirms=true spring.rabbitmq.publisher-returns=true spring.rabbitmq.template.mandatory=true
第三步：配置OrderQueueConfig
packagecom.tuohang.platform.config;
importorg.springframework.amqp.core.Binding;
importorg.springframework.amqp.core.BindingBuilder;
importorg.springframework.amqp.core.DirectExchange;
importorg.springframework.amqp.core.Queue;
importorg.springframework.amqp.core.QueueBuilder;
importorg.springframework.amqp.rabbit.connection.ConnectionFactory;
importorg.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
importorg.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
importorg.springframework.context.annotation.Bean;
importorg.springframework.context.annotation.Configuration;
/**
*rabbitMQ的队列设置（生产者发送的消息，永远是先进入exchange，再通过路由，转发到队列）
*
*
*@authorAdministrator
*@version1.0
*@Date2018年9月18日
*/
@Configuration
publicclassOrderQueueConfig{
/**
*订单缓冲交换机名称
*/
publicfinalstaticStringORDER_PRE_EXCHANGE_NAME="order_pre_exchange";
/**
*发送到该队列的message会在一段时间后过期进入到order_delay_process_queue【队列里所有的message都有统一的失效时间】
*/
publicfinalstaticStringORDER_PRE_TTL_DELAY_QUEUE_NAME="order_pre_ttl_delay_queue";
/**
*订单的交换机DLX名字
*/
finalstaticStringORDER_DELAY_EXCHANGE_NAME="order_delay_exchange";
/**
*订单message时间过期后进入的队列，也就是订单实际的消费队列
*/
publicfinalstaticStringORDER_DELAY_PROCESS_QUEUE_NAME="order_delay_process_queue";
/**
*订单在缓冲队列过期时间（毫秒）30分钟
*/
publicfinalstaticintORDER_QUEUE_EXPIRATION=1800000;
/**
*订单缓冲交换机
*
*@return
*/
@Bean
publicDirectExchangepreOrderExange(){
returnnewDirectExchange(ORDER_PRE_EXCHANGE_NAME);
}
/**
*创建order_per_ttl_delay_queue队列，订单消息经过缓冲交换机，会进入该队列
*
*@return
*/
@Bean
publicQueuedelayQueuePerOrderTTLQueue(){
returnQueueBuilder.durable(ORDER_PRE_TTL_DELAY_QUEUE_NAME)
.withArgument("x-dead-letter-exchange",ORDER_DELAY_EXCHANGE_NAME)//DLX
.withArgument("x-dead-letter-routing-key",ORDER_DELAY_PROCESS_QUEUE_NAME)//deadletter携带的routingkey
.withArgument("x-message-ttl",ORDER_QUEUE_EXPIRATION)//设置订单队列的过期时间
.build();
}
/**
*将order_pre_exchange绑定到order_pre_ttl_delay_queue队列
*
*@paramdelayQueuePerOrderTTLQueue
*@parampreOrderExange
*@return
*/
@Bean
publicBindingqueueOrderTTLBinding(QueuedelayQueuePerOrderTTLQueue,DirectExchangepreOrderExange){
returnBindingBuilder.bind(delayQueuePerOrderTTLQueue).to(preOrderExange).with(ORDER_PRE_TTL_DELAY_QUEUE_NAME);
}
/**
*创建订单的DLXexchange
*
*@return
*/
@Bean
publicDirectExchangedelayOrderExchange(){
returnnewDirectExchange(ORDER_DELAY_EXCHANGE_NAME);
}
/**
*创建order_delay_process_queue队列，也就是订单实际消费队列
*
*@return
*/
@Bean
publicQueuedelayProcessOrderQueue(){
returnQueueBuilder.durable(ORDER_DELAY_PROCESS_QUEUE_NAME).build();
}
/**
*将DLX绑定到实际消费队列
*
*@paramdelayProcessOrderQueue
*@paramdelayExchange
*@return
*/
@Bean
publicBindingdlxOrderBinding(QueuedelayProcessOrderQueue,DirectExchangedelayOrderExchange){
returnBindingBuilder.bind(delayProcessOrderQueue).to(delayOrderExchange).with(ORDER_DELAY_PROCESS_QUEUE_NAME);
}
/**
*监听订单实际消费者队列order_delay_process_queue
*
*@paramconnectionFactory
*@paramprocessReceiver
*@return
*/
@Bean
publicSimpleMessageListenerContainerorderProcessContainer(ConnectionFactoryconnectionFactory,
OrderProcessReceiverprocessReceiver){
SimpleMessageListenerContainercontainer=newSimpleMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
container.setQueueNames(ORDER_DELAY_PROCESS_QUEUE_NAME);//监听order_delay_process_queue
container.setMessageListener(newMessageListenerAdapter(processReceiver));
returncontainer;
}
}
消费者OrderProcessReceiver：
packagecom.tuohang.platform.config;
importjava.util.Objects;
importorg.apache.tools.ant.types.resources.selectors.Date;
importorg.slf4j.Logger;
importorg.slf4j.LoggerFactory;
importorg.springframework.amqp.core.Message;
importorg.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
importorg.springframework.stereotype.Component;
importcom.rabbitmq.client.Channel;
/**
*订单延迟处理消费者
*
*
*@authorAdministrator
*@version1.0
*@Date2018年9月18日
*/
@Component
publicclassOrderProcessReceiverimplementsChannelAwareMessageListener{
privatestaticLoggerlogger=LoggerFactory.getLogger(OrderProcessReceiver.class);
Stringmsg="Thefailedmessagewillautoretryafteracertaindelay";
@Override
publicvoidonMessage(Messagemessage,Channelchannel)throwsException{
try{
processMessage(message);
}catch(Exceptione){
//如果发生了异常，则将该消息重定向到缓冲队列，会在一定延迟之后自动重做
channel.basicPublish(OrderQueueConfig.ORDER_PRE_EXCHANGE_NAME,OrderQueueConfig.ORDER_PRE_TTL_DELAY_QUEUE_NAME,null,
msg.getBytes());
}
}
/**
*处理订单消息，如果订单未支付，取消订单（如果当消息内容为FAIL_MESSAGE的话，则需要抛出异常）
*
*@parammessage
*@throwsException
*/
publicvoidprocessMessage(Messagemessage)throwsException{
StringrealMessage=newString(message.getBody());
logger.info("Received<"+realMessage+">");
//取消订单
if(!Objects.equals(realMessage,msg)){
//SpringKit.getBean(ITestService.class).resetSexById(Long.valueOf(realMessage));
System.out.println("测试111111-----------"+newDate());
System.out.println(message);
}
}
}
或者
/**
*测试rabbit消费者
*
*
*@authorAdministrator
*@version1.0
*@Date2018年9月25日
*/
@Component
@RabbitListener(queues=TestQueueConfig.TEST_DELAY_PROCESS_QUEUE_NAME)
publicclassTestProcessReceiver{
privatestaticLoggerlogger=LoggerFactory.getLogger(TestProcessReceiver.class);
Stringmsg="Thefailedmessagewillautoretryafteracertaindelay";
@RabbitHandler
publicvoidonMessage(Messagemessage,Channelchannel)throwsException{
try{
processMessage(message);
//告诉服务器收到这条消息已经被我消费了可以在队列删掉；否则消息服务器以为这条消息没处理掉后续还会在发
channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
}catch(Exceptione){
//如果发生了异常，则将该消息重定向到缓冲队列，会在一定延迟之后自动重做
channel.basicPublish(TestQueueConfig.TEST_PRE_EXCHANGE_NAME,TestQueueConfig.TEST_PRE_TTL_DELAY_QUEUE_NAME,null,
msg.getBytes());
}
}
/**
*处理订单消息，如果订单未支付，取消订单（如果当消息内容为FAIL_MESSAGE的话，则需要抛出异常）
*
*@parammessage
*@throwsException
*/
publicvoidprocessMessage(Messagemessage)throwsException{
StringrealMessage=newString(message.getBody());
logger.info("Received<"+realMessage+">");
//取消订单
if(!Objects.equals(realMessage,msg)){
System.out.println("测试111111-----------"+newDate());
}else{
System.out.println("rabbitelse...");
}
}
}
生产者
/**
*测试rabbitmq
*
*@return
*/
@RequestMapping(value="/testrab")
publicStringtestraa(){
GenericResultgr=null;
try{
Stringname="test_pre_ttl_delay_queue";
longexpiration=10000;//10s过期时间
rabbitTemplate.convertAndSend(name,String.valueOf(123456));
//在单个消息上设置过期时间
//rabbitTemplate.convertAndSend(name,(Object)String.valueOf(123456),newExpirationMessagePostProcessor(expiration));
}catch(ServiceExceptione){
e.printStackTrace();
gr=newGenericResult(StateCode.ERROR,languageMap.get("network_error"),e.getMessage());
}
returngetWrite(gr);
}
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持毛票票。
声明：本文内容来源于网络，版权归原作者所有，内容由互联网用户自发贡献自行上传，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任。如果您发现有涉嫌版权的内容，欢迎发送邮件至：czq8825#qq.com（发邮件时，请将#更换为@）进行举报，并提供相关证据，一经查实，本站将立刻删除涉嫌侵权内容。