使用 RabbitMq

如何使用

• 安装RabbitMq服务

安装 Eralng OTP ，安装 Rabbit MQ Server，具体可以到网上搜教程。

• spring 集成 RabbitMq

spring配置文件：

<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:rabbit="http://www.springframework.org/schema/rabbit" 
    xsi:schemaLocation="http://www.springframework.org/schema/beans 
    http://www.springframework.org/schema/beans/spring-beans-3.0.xsd 
    http://www.springframework.org/schema/rabbit
    http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd" >
    <description>rabbitmq 连接服务配置</description>
    <!-- 连接配置 -->
    <rabbit:connection-factory id="connectionFactory" host="${mq.host}" 
        username="${mq.username}" password="${mq.password}" port="${mq.port}"/> 
    <!--
        通过指定下面的admin信息，
        当前producer中的exchange和queue会在rabbitmq服务器上自动生成 
    -->
    <rabbit:admin connection-factory="connectionFactory"/>
    <!-- spring template声明-->
    <rabbit:template id="amqpTemplate" exchange="sms_exchange" 
        connection-factory="connectionFactory" message-converter="jsonMessageConverter" /> 
    <!-- 消息对象json转换类 -->
    <bean id="jsonMessageConverter" 
        class="org.springframework.amqp.support.converter.Jackson2JsonMessageConverter" /> 
    <!-- 申明一个消息队列Queue ; 
        durable:是否持久化 ; 
        exclusive: 仅创建者可以使用的私有队列，断开后自动删除; 
        auto_delete: 当所有消费客户端连接断开后,是否自动删除队列;
    -->
    <rabbit:queue name="sms_code_queue" durable="true" auto-delete="false" 
        exclusive="false" />
    <!-- 交换机定义 (
        rabbit:direct-exchange：
            定义exchange模式为direct，意思就是消息与一个特定的路由键完全匹配，才会转发。 
        rabbit:binding：设置消息queue匹配的key)
    -->
    <rabbit:direct-exchange  name="sms_exchange" durable="true" auto-delete="false" >
        <rabbit:bindings>
            <rabbit:binding queue="sms_code_queue" key="sms_code_queue_key"/>
        </rabbit:bindings>
    </rabbit:direct-exchange>
    <!-- 监听配置 (queues：监听的队列，多个的话用逗号（,）分隔ref：监听器) -->
    <rabbit:listener-container connection-factory="connectionFactory" acknowledge="auto">
        <rabbit:listener queues="sms_code_queue" ref="smsCodeQueueListenter"/>
    </rabbit:listener-container>    
</beans>

消息推送：

@Service
public class SmsMQProducerImpl implements MQProducer{
    @Autowired
    private AmqpTemplate amqpTemplate;
    private final static Logger logger = Logger.getLogger(SmsMQProducerImpl.class);
    @Override
    public void sendDataToQueue(String key, Object object) {
        try {
            amqpTemplate.convertAndSend("sms_code_queue_key", object);
        } catch (Exception e) {
            logger.error(e);
        }
    }
}

消息监听：

@Component
public class SmsCodeQueueListenter implements MessageListener {
    @Override
    public void onMessage(Message message) {

    }
}

简单解释

<rabbit:connection-factory />：连接RabbitMq服务

<rabbit:queue />：定义队列

<rabbit:direct-exchange />：定义交换机，关联队列，消息将直接推送给交换机，交换机将消息转发至队列

<rabbit:listener-container />：指定队列的listener

<rabbit:template/>：负责向队列推送消息的bean，关联交换机

<rabbit:admin/>：自动在RabbitMq服务器上生成定义的队列和交换机