http://blog.csdn.net/zhu_tianwei/article/details/40919249

實(shí)現(xiàn)使用Exchange類型為DirectExchange. routingkey的名稱默認(rèn)為Queue的名稱。注解實(shí)現(xiàn)異步發(fā)送消息。

1.生產(chǎn)者配置ProducerConfiguration.Java

?

[java]?view plaincopy print?

package?cn.slimsmart.rabbitmq.demo.spring.async;??

??

import?java.util.concurrent.atomic.AtomicInteger;??

??

import?org.springframework.amqp.rabbit.connection.CachingConnectionFactory;??

import?org.springframework.amqp.rabbit.connection.ConnectionFactory;??

import?org.springframework.amqp.rabbit.core.RabbitTemplate;??

import?org.springframework.beans.factory.annotation.Autowired;??

import?org.springframework.beans.factory.config.BeanPostProcessor;??

import?org.springframework.context.annotation.Bean;??

import?org.springframework.context.annotation.Configuration;??

import?org.springframework.scheduling.annotation.Scheduled;??

import?org.springframework.scheduling.annotation.ScheduledAnnotationBeanPostProcessor;??

??

import?com.rabbitmq.client.AMQP;??

??

@Configuration??

public?class?ProducerConfiguration?{??

??

????//?指定隊(duì)列名稱?routingkey的名稱默認(rèn)為Queue的名稱，使用Exchange類型為DirectExchange??

????protected?final?String?helloWorldQueueName?=?"spring-queue-async";??

??

????//?創(chuàng)建鏈接??

????@Bean??

????public?ConnectionFactory?connectionFactory()?{??

????????CachingConnectionFactory?connectionFactory?=?new?CachingConnectionFactory("192.168.36.102");??

????????connectionFactory.setUsername("admin");??

????????connectionFactory.setPassword("admin");??

????????connectionFactory.setPort(AMQP.PROTOCOL.PORT);??

????????return?connectionFactory;??

????}??

??

????//?創(chuàng)建rabbitTemplate?消息模板類??

????@Bean??

????public?RabbitTemplate?rabbitTemplate()?{??

????????RabbitTemplate?template?=?new?RabbitTemplate(connectionFactory());??

????????template.setRoutingKey(this.helloWorldQueueName);??

????????return?template;??

????}??

??

????//創(chuàng)建一個(gè)調(diào)度??

????@Bean??

????public?ScheduledProducer?scheduledProducer()?{??

????????return?new?ScheduledProducer();??

????}??

??

????@Bean??

????public?BeanPostProcessor?postProcessor()?{??

????????return?new?ScheduledAnnotationBeanPostProcessor();??

????}??

??

??????

????static?class?ScheduledProducer?{??

??

????????@Autowired??

????????private?volatile?RabbitTemplate?rabbitTemplate;??

??

????????//自增整數(shù)??

????????private?final?AtomicInteger?counter?=?new?AtomicInteger();??

????????/**?

?????????*?每3秒發(fā)送一條消息?

?????????*??

?????????*?Spring3中加強(qiáng)了注解的使用，其中計(jì)劃任務(wù)也得到了增強(qiáng)，現(xiàn)在創(chuàng)建一個(gè)計(jì)劃任務(wù)只需要兩步就完成了：?

????????創(chuàng)建一個(gè)Java類，添加一個(gè)無(wú)參無(wú)返回值的方法，在方法上用@Scheduled注解修飾一下；?

????????在Spring配置文件中添加三個(gè)<task:****?/>節(jié)點(diǎn)；?

????????參考：http://zywang.iteye.com/blog/949123?

?????????*/??

????????@Scheduled(fixedRate?=?3000)??

????????public?void?sendMessage()?{??

????????????rabbitTemplate.convertAndSend("Hello?World?"?+?counter.incrementAndGet());??

????????}??

????}??

??

}??

2.生產(chǎn)者啟動(dòng)類Producer,java

?

?

[java]?view plaincopy print?

package?cn.slimsmart.rabbitmq.demo.spring.async;??

??

import?org.springframework.context.annotation.AnnotationConfigApplicationContext;??

??

public?class?Producer?{??

??

????public?static?void?main(String[]?args)?{??

????????new?AnnotationConfigApplicationContext(ProducerConfiguration.class);??

????}??

}??

3.接收消息處理類ReceiveMsgHandler.java

?

?

[java]?view plaincopy print?

package?cn.slimsmart.rabbitmq.demo.spring.async;??

??

public?class?ReceiveMsgHandler?{??

??

????public?void?handleMessage(String?text)?{??

????????System.out.println("Received:?"?+?text);??

????}??

}??

4.消費(fèi)者配置ConsumerConfiguration

?

?

[java]?view plaincopy print?

package?cn.slimsmart.rabbitmq.demo.spring.async;??

??

import?org.springframework.amqp.core.AmqpAdmin;??

import?org.springframework.amqp.core.Queue;??

import?org.springframework.amqp.rabbit.connection.CachingConnectionFactory;??

import?org.springframework.amqp.rabbit.connection.ConnectionFactory;??

import?org.springframework.amqp.rabbit.core.RabbitAdmin;??

import?org.springframework.amqp.rabbit.core.RabbitTemplate;??

import?org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;??

import?org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;??

import?org.springframework.context.annotation.Bean;??

import?org.springframework.context.annotation.Configuration;??

??

import?com.rabbitmq.client.AMQP;??

??

@Configuration??

public?class?ConsumerConfiguration?{??

??

????//?指定隊(duì)列名稱?routingkey的名稱默認(rèn)為Queue的名稱，使用Exchange類型為DirectExchange??

????protected?String?springQueueDemo?=?"spring-queue-async";??

??

????//?創(chuàng)建鏈接??

????@Bean??

????public?ConnectionFactory?connectionFactory()?{??

????????CachingConnectionFactory?connectionFactory?=?new?CachingConnectionFactory(??

????????????????"192.168.36.102");??

????????connectionFactory.setUsername("admin");??

????????connectionFactory.setPassword("admin");??

????????connectionFactory.setPort(AMQP.PROTOCOL.PORT);??

????????return?connectionFactory;??

????}??

??

????//?創(chuàng)建rabbitAdmin?代理類??

????@Bean??

????public?AmqpAdmin?amqpAdmin()?{??

????????return?new?RabbitAdmin(connectionFactory());??

????}??

??

????//?創(chuàng)建rabbitTemplate?消息模板類??

????@Bean??

????public?RabbitTemplate?rabbitTemplate()?{??

????????RabbitTemplate?template?=?new?RabbitTemplate(connectionFactory());??

????????//?The?routing?key?is?set?to?the?name?of?the?queue?by?the?broker?for?the??

????????//?default?exchange.??

????????template.setRoutingKey(this.springQueueDemo);??

????????//?Where?we?will?synchronously?receive?messages?from??

????????template.setQueue(this.springQueueDemo);??

????????return?template;??

????}??

??

????//??

????//?Every?queue?is?bound?to?the?default?direct?exchange??

????public?Queue?helloWorldQueue()?{??

????????return?new?Queue(this.springQueueDemo);??

????}??

??

????@Bean??

????public?SimpleMessageListenerContainer?listenerContainer()?{??

????????SimpleMessageListenerContainer?container?=?new?SimpleMessageListenerContainer();??

????????container.setConnectionFactory(connectionFactory());??

????????container.setQueueNames(this.springQueueDemo);??

????????container.setMessageListener(new?MessageListenerAdapter(??

????????????????new?ReceiveMsgHandler()));??

????????return?container;??

????}??

??

}??

5.消費(fèi)者啟動(dòng)類Consumer.java

?

?

[java]?view plaincopy print?

package?cn.slimsmart.rabbitmq.demo.spring.async;??

??

import?org.springframework.context.annotation.AnnotationConfigApplicationContext;??

??

public?class?Consumer?{??

????public?static?void?main(String[]?args)?{??

????????new?AnnotationConfigApplicationContext(ConsumerConfiguration.class);??

????}??

}??

啟動(dòng)接收消息，再發(fā)送消息

?

?

[sql]?view plaincopy print?

Received:?Hello?World?1??

Received:?Hello?World?2??

Received:?Hello?World?3??

Received:?Hello?World?4??

Received:?Hello?World?5??

Received:?Hello?World?6??

Received:?Hello?World?7??

......??

?

若報(bào)spring-queue-async消息隊(duì)列不存在，請(qǐng)?jiān)诳刂婆_(tái)添加。

總結(jié)

以上是生活随笔為你收集整理的（转）RabbitMQ学习之spring整合发送异步消息(注解实现)的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

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