RabbitMQ的使用

提供者

pom.xml 导入依赖jar包

applicationContext.xml 文件中导入rabbitMQ

 <import resource="classpath:spring-rabbit.xml"/> 

 spring-rabbit.xml 文件

 1.连接工厂

<!-- supply vhost 连接工厂 -->
    <rabbit:connection-factory
            id="itemConnectionFactory"
            addresses="${mq.addresses}"
            virtual-host="${mq.vhostSupply}"
            username="${mq.username}"
            password="${mq.password}"
            cache-mode="${mq.cacheMode}"
            channel-cache-size="${mq.cacheSize}"
            connection-timeout="${mq.connectionTimeout}"
    />

 MQ的配置

# RabbitMQ
mq.host=${filter.mq.host}
mq.port=${filter.mq.port}
mq.addresses=${filter.mq.addresses}
mq.username=${filter.mq.username}
mq.password=${filter.mq.password}
mq.vhostSupply=${filter.mq.vhostSupply}
mq.cacheMode=${filter.mq.cacheMode}
mq.cacheSize=${filter.mq.cacheSize}
mq.connectionTimeout=${filter.mq.connectionTimeout}
mq.requestedHeartbeat=${filter.mq.requestedHeartbeat}


# RabbitMQ
filter.mq.host=
filter.mq.port=
filter.mq.addresses=127.0.0.1:5672
filter.mq.username=dev
filter.mq.password=U69o3NGHtZAmfLX5dQGW
filter.mq.vhostSupply=supply
filter.mq.cacheMode=CHANNEL
filter.mq.cacheSize=10
filter.mq.connectionTimeout=3000
filter.mq.requestedHeartbeat=

2.vhost 管理契

 <!-- item vhost 管理器 --> 2 <rabbit:admin id="itemAdmin" connection-factory="itemConnectionFactory"/> 

3.exchange

    <rabbit:topic-exchange name="exchange.item.add.Original" declared-by="itemAdmin" auto-declare="true" auto-delete="false" durable="true">
        <rabbit:exchange-arguments>
        </rabbit:exchange-arguments>
        <rabbit:bindings>
            <rabbit:binding pattern="soa.item.originalItem.created" queue="soa.item.originalItem.created.queue" />
        </rabbit:bindings>
    </rabbit:topic-exchange>

4.queue

<rabbit:queue name="soa.item.originalItem.created.queue" declared-by="itemAdmin" auto-declare="true" auto-delete="false" durable="true" exclusive="false">
        <rabbit:queue-arguments>
            <entry key="x-max-length" value="1000000" value-type="java.lang.Long"/>
            <entry key="x-max-length-bytes" value="300485760" value-type="java.lang.Long"/>
        </rabbit:queue-arguments>
    </rabbit:queue>

5.template

    <!-- JSON 消息转换器 -->
    <bean id="jsonMessageConvertor" class="com.lcb.soa.misc.common.mq.FastJsonMessageConverter"/>
    
    <rabbit:template id="itemEventTemplate" connection-factory="itemConnectionFactory"
                     exchange="exchange.item.add.Original" message-converter="jsonMessageConvertor"/>
    
    <bean id="originalItemEventMqSender" class="com.lcb.soa.misc.common.mq.MQSender" p:template-ref="itemEventTemplate"/>

6.消息发送

    @Resource(name = "originalItemEventMqSender")
    private MQSender originalItemEventMqSender;

4     originalItemEventMqSender.sendAsync("soa.item.originalItem.created", addOriginalItemNotify);

>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>　　util　　>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>

package com.lcb.soa.misc.common.mq;

import java.io.UnsupportedEncodingException;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.support.converter.AbstractMessageConverter;
import org.springframework.amqp.support.converter.MessageConversionException;

import com.alibaba.fastjson.JSON;

/** JSON 消息转换器 */
public class FastJsonMessageConverter extends AbstractMessageConverter
{
    public static final String DEFAULT_CHARSET = "UTF-8";
    private String defaultCharset = DEFAULT_CHARSET;

    @Override
    protected Message createMessage(Object obj, MessageProperties props)
    {
        try
        {
            String json = JSON.toJSONString(obj);
            byte[] body = json.getBytes(defaultCharset);

            props.setContentType(MessageProperties.CONTENT_TYPE_JSON);
            props.setContentEncoding(defaultCharset);

            if(body != null)
                props.setContentLength(body.length);

            return new Message(body, props);
        }
        catch(UnsupportedEncodingException e)
        {
            throw new MessageConversionException("invalid charset " + defaultCharset, e);
        }
    }

    @Override
    public Object fromMessage(Message message)
    {
        String charset = message.getMessageProperties().getContentEncoding();
        
        if(charset == null)
            charset = defaultCharset;

        try
        {
            return new String(message.getBody(), charset);
        }
        catch(UnsupportedEncodingException e)
        {
            throw new MessageConversionException("invalid charset " + charset, e);
        }
    }

    public void setDefaultCharset(String defaultCharset)
    {
        this.defaultCharset = (defaultCharset != null ? defaultCharset : "UTF-8");
    }

    public String getDefaultCharset()
    {
        return this.defaultCharset;
    }

}

package com.lcb.soa.misc.common.mq;

import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.scheduling.annotation.Async;

/** MQ 消息发送器 */
public class MQSender
{
    private AmqpTemplate template;
    
    public void send(Object message) throws AmqpException
    {
        template.convertAndSend(message);
    }
    
    public void send(String routingKey, Object message) throws AmqpException
    {
        template.convertAndSend(routingKey, message);
    }
    
    public void send(String exchange, String routingKey, Object message) throws AmqpException
    {
        template.convertAndSend(exchange, routingKey, message);
    }
    
    public void send(Object message, MessagePostProcessor messagePostProcessor) throws AmqpException
    {
        template.convertAndSend(message, messagePostProcessor);
    }
    
    public void send(String routingKey, Object message, MessagePostProcessor messagePostProcessor) throws AmqpException
    {
        template.convertAndSend(routingKey, message, messagePostProcessor);
    }
    
    public void send(String exchange, String routingKey, Object message, MessagePostProcessor messagePostProcessor) throws AmqpException
    {
        template.convertAndSend(exchange, routingKey, message, messagePostProcessor);
    }

    @Async
    public void sendAsync(Object message) throws AmqpException
    {
        template.convertAndSend(message);
    }
    
    @Async
    public void sendAsync(String routingKey, Object message) throws AmqpException
    {
        template.convertAndSend(routingKey, message);
    }
    
    @Async
    public void sendAsync(String exchange, String routingKey, Object message) throws AmqpException
    {
        template.convertAndSend(exchange, routingKey, message);
    }
    
    @Async
    public void sendAsync(Object message, MessagePostProcessor messagePostProcessor) throws AmqpException
    {
        template.convertAndSend(message, messagePostProcessor);
    }
    
    @Async
    public void sendAsync(String routingKey, Object message, MessagePostProcessor messagePostProcessor) throws AmqpException
    {
        template.convertAndSend(routingKey, message, messagePostProcessor);
    }
    
    @Async
    public void sendAsync(String exchange, String routingKey, Object message, MessagePostProcessor messagePostProcessor) throws AmqpException
    {
        template.convertAndSend(exchange, routingKey, message, messagePostProcessor);
    }

    public AmqpTemplate getTemplate()
    {
        return template;
    }

    public void setTemplate(AmqpTemplate template)
    {
        this.template = template;
    }

}

消费者

1.连接工厂，省略

2.listener

    <!-- 原厂件 商品添加  -->
    <rabbit:listener-container connection-factory="supplyConnectionFactory" acknowledge="auto" message-converter="jsonMessageConvertor">
        <rabbit:listener queue-names="soa.item.originalItem.created.queue" ref="itemOriginalListener" method="onAddOriginalItemMessage"/>
    </rabbit:listener-container>
    <bean id="sprayOrderSmsListener" class="com.lcb.listener.item.ItemOriginalListener"/>

3.class

@Component
public class ItemOriginalListener {
    private static Logger logger = LoggerFactory.getLogger(ItemOriginalListener.class);
    @Autowired
    private IItemOriginalInfoService itemOriginalInfoService;

    public void onAddOriginalItemMessage(String json) {

    }
}