Spring集成RabbitMq

　　　　　com.fasterxml.jackson.corejackson-databind2.7.5org.springframework.amqpspring-rabbit2.1.7.RELEASE

xmlns:xsi&＃61;"http://www.w3.org/2001/XMLSchema-instance" xmlns:tx&＃61;"http://www.springframework.org/schema/tx"xmlns:task&＃61;"http://www.springframework.org/schema/task" xmlns:context&＃61;"http://www.springframework.org/schema/context"xmlns:aop&＃61;"http://www.springframework.org/schema/aop"xsi:schemaLocation&＃61;"http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.0.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-3.0.xsd http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.0.xsd">package&＃61;"com.pinghengxing..*"><import resource&＃61;"classpath*:com/config/rabbitmq_producer.xml" /><import resource&＃61;"classpath*:com/config/rabbitmq_consumer.xml" />

xmlns:xsi&＃61;"http://www.w3.org/2001/XMLSchema-instance" xmlns:context&＃61;"http://www.springframework.org/schema/context"xmlns:rabbit&＃61;"http://www.springframework.org/schema/rabbit"xsi:schemaLocation&＃61;"http://www.springframework.org/schema/beanshttp://www.springframework.org/schema/beans/spring-beans.xsdhttp://www.springframework.org/schema/contexthttp://www.springframework.org/schema/context/spring-context-4.3.xsdhttp://www.springframework.org/schema/rabbithttp://www.springframework.org/schema/rabbit/spring-rabbit.xsd">host&＃61;"140.143.xx.xx" username&＃61;"ww" password&＃61;"ww" port&＃61;"5672"virtual-host&＃61;"ww" channel-cache-size&＃61;"25" cache-mode&＃61;"CHANNEL"publisher-confirms&＃61;"true" publisher-returns&＃61;"true" connection-timeout&＃61;"200" />class&＃61;"org.springframework.retry.support.RetryTemplate">class&＃61;"org.springframework.retry.backoff.ExponentialBackOffPolicy">class&＃61;"org.springframework.retry.policy.SimpleRetryPolicy"> message-converter&＃61;"jsonMessageConverter" confirm-callback&＃61;"confirmCallback" return-callback&＃61;"returnCallback" mandatory&＃61;"true" />

class&＃61;"org.springframework.amqp.support.converter.Jackson2JsonMessageConverter">

xmlns:xsi&＃61;"http://www.w3.org/2001/XMLSchema-instance" xmlns:context&＃61;"http://www.springframework.org/schema/context"xmlns:rabbit&＃61;"http://www.springframework.org/schema/rabbit"xsi:schemaLocation&＃61;"http://www.springframework.org/schema/beanshttp://www.springframework.org/schema/beans/spring-beans.xsdhttp://www.springframework.org/schema/contexthttp://www.springframework.org/schema/context/spring-context-4.3.xsdhttp://www.springframework.org/schema/rabbithttp://www.springframework.org/schema/rabbit/spring-rabbit.xsd">host&＃61;"140.143.xx.xx" username&＃61;"ww" password&＃61;"ww" port&＃61;"5672"virtual-host&＃61;"ww" channel-cache-size&＃61;"25" cache-mode&＃61;"CHANNEL"connection-timeout&＃61;"200" />prefetch&＃61;"1" concurrency&＃61;"1" >

package com.pinghengxing.direct;import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;/** * &＃64;author ww*/public class DirectProducer {private static ApplicationContext ac;public static void sendMessage(String exchange,String routingKey,Object message){ac &＃61; new ClassPathXmlApplicationContext("classpath:com/config/springContext.xml");RabbitTemplate rt &＃61; ac.getBean(RabbitTemplate.class);for(int i&＃61;0;i<10;i&＃43;&＃43;){rt.convertAndSend(exchange, routingKey, message&＃43;""&＃43;i);}}public static void main(String[] args) {DirectProducer.sendMessage("myDirectExchange","direct","路由模式");}}

package com.pinghengxing.direct;import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;import com.rabbitmq.client.Channel;
&＃64;Component("directReceiver")
public class DirectReceiver implements ChannelAwareMessageListener{&＃64;Overridepublic void onMessage(Message message, Channel channel) throws Exception {System.out.println("************************direct111********************************");System.out.println("路由模式direct111 接收信息&＃xff1a;"&＃43;new String(message.getBody()));System.out.println("********************************************************");//设置手工应答
// if(true){
// throw new Exception();
// }channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);}}

package com.pinghengxing.direct;import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;import com.rabbitmq.client.Channel;
&＃64;Component("directReceiver2")
public class DirectReceiver2 implements ChannelAwareMessageListener{&＃64;Overridepublic void onMessage(Message message, Channel channel) throws Exception {System.out.println("************************direct222********************************");System.out.println("路由模式direct222 接收信息&＃xff1a;"&＃43;new String(message.getBody()));System.out.println("********************************************************");//设置手工应答
// if(true){
// throw new Exception();
// }channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);}}

host&＃61;"140.143.xx.xx" username&＃61;"ww" password&＃61;"ww" port&＃61;"5672"virtual-host&＃61;"ww" channel-cache-size&＃61;"25" cache-mode&＃61;"CHANNEL"publisher-confirms&＃61;"true" publisher-returns&＃61;"true" connection-timeout&＃61;"200" />

message-converter&＃61;"jsonMessageConverter" confirm-callback&＃61;"confirmCallback" return-callback&＃61;"returnCallback" mandatory&＃61;"true" />

package com.pinghengxing.callback;import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;/*** confirm-callback监听&＃xff08;用于监听exchange是否接收成功&＃xff09;* &＃64;author ww**/
&＃64;Component("confirmCallback")
public class ConfirmCallBack implements RabbitTemplate.ConfirmCallback{/*** CorrelationData 是在发送消息时传入回调方法的参数&＃xff0c;可以用于区分消息对象。 CorrelationData对象中只有一个属性 String id。* 通过这个参数&＃xff0c;我们可以区分当前是发送哪一条消息时的回调&＃xff0c;并通过ack参数来进行失败重发功能* &＃64;param correlationData 回调的相关数据.* &＃64;param ack true for ack, false for nack* &＃64;param cause 专门给NACK准备的一个可选的原因&＃xff0c;其他情况为null。*/public void confirm(CorrelationData correlationData, boolean ack,String cause) {System.out.println("********************************************************");System.out.println("exChange确认" &＃43; ack &＃43; "   " &＃43; cause);System.out.println("********************************************************");}}

host&＃61;"140.143.xx.xx" username&＃61;"ww" password&＃61;"ww" port&＃61;"5672"
virtual-host&＃61;"ww" channel-cache-size&＃61;"25" cache-mode&＃61;"CHANNEL"
publisher-confirms&＃61;"true" publisher-returns&＃61;"true" connection-timeout&＃61;"200" />

message-converter&＃61;"jsonMessageConverter" confirm-callback&＃61;"confirmCallback" return-callback&＃61;"returnCallback" mandatory&＃61;"true" />

package com.pinghengxing.callback;import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;/*** 推送消息到queue失败时回调* &＃64;author ww**/
&＃64;Component("returnCallback")
public class ReturnCallBack implements RabbitTemplate.ReturnCallback {&＃64;Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {System.out.println("********************************************************");System.out.println("失败确认:"&＃43;message&＃43;" | "&＃43;replyCode&＃43;" | "&＃43;replyText&＃43;" | "&＃43;exchange&＃43;" | "&＃43;routingKey);System.out.println("********************************************************");}}

　　　　 com.fasterxml.jackson.corejackson-databind2.7.5

class&＃61;"org.springframework.amqp.support.converter.Jackson2JsonMessageConverter">

message-converter&＃61;"jsonMessageConverter" confirm-callback&＃61;"confirmCallback" return-callback&＃61;"returnCallback" mandatory&＃61;"true" />