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

實現使用Exchange類型為DirectExchange. routingkey的名稱默認為Queue的名稱。注解實現異步發送消息。

1.生產者配置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?{??

??

????//?指定隊列名稱?routingkey的名稱默認為Queue的名稱，使用Exchange類型為DirectExchange??

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

??

????//?創建鏈接??

????@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;??

????}??

??

????//?創建rabbitTemplate?消息模板類??

????@Bean??

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

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

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

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

????}??

??

????//創建一個調度??

????@Bean??

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

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

????}??

??

????@Bean??

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

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

????}??

??

??????

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

??

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

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

??

????????//自增整數??

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

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

?????????*?每3秒發送一條消息?

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

?????????*?Spring3中加強了注解的使用，其中計劃任務也得到了增強，現在創建一個計劃任務只需要兩步就完成了：?

????????創建一個Java類，添加一個無參無返回值的方法，在方法上用@Scheduled注解修飾一下；?

????????在Spring配置文件中添加三個<task:****?/>節點；?

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

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

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

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

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

????????}??

????}??

??

}??

2.生產者啟動類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.消費者配置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?{??

??

????//?指定隊列名稱?routingkey的名稱默認為Queue的名稱，使用Exchange類型為DirectExchange??

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

??

????//?創建鏈接??

????@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;??

????}??

??

????//?創建rabbitAdmin?代理類??

????@Bean??

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

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

????}??

??

????//?創建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.消費者啟動類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);??

????}??

}??

啟動接收消息，再發送消息

?

?

[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??

......??

?

若報spring-queue-async消息隊列不存在，請在控制臺添加。

總結

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

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。