springmvc 整合rabbitmq 配置_rabbitmq topic springmvc设置

1.springmvc配置

<?xml version="1.0" encoding="UTF-8"?>
<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}" virtual-host="${mq.vhost}"/>
    <rabbit:admin connection-factory="connectionFactory"/>
    <!-- 定义模板 -->
    <rabbit:template exchange="topic.exchange" id="amqpTemplate" connection-factory="connectionFactory"/>
    <!-- 定义交换机 -->
    <rabbit:topic-exchange name="topic.exchange" auto-delete="true"/>

    <!--  申明一个消息队列Queue   -->
    <rabbit:queue id="rabbitQueue" name="${mq.cluster.queue}" durable="true" auto-delete="true"/>
    <bean id="serverMessageListener" class="com.hh.listener.ServerMessageListener"/>
    <bean id="messageContainer" class="org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer">
        <property name="queues" ref="rabbitQueue"></property>
        <property name="exposeListenerChannel" value="true"></property>
        <property name="maxConcurrentConsumers" value="10"></property>
        <property name="concurrentConsumers" value="2"></property>
        <property name="connectionFactory" ref="connectionFactory"></property>
        <property name="acknowledgeMode" value="MANUAL"></property>
        <property name="messageListener" ref="serverMessageListener"></property>
    </bean>
</beans>

2.监听类

@Slf4j
@Component
public class ServerMessageListener implements ChannelAwareMessageListener {
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void onMessage(Message message, Channel channel) {
        String msg = new String(message.getBody());
        log.info("消费{}", msg);
        try {
            //消息的标识，false只确认当前一个消息收到，true确认所有consumer获得的消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

3.生产类

@Resource
private AmqpTemplate amqpTemplate;
private void sendDataToQueue(String routingKey, Object object) {
    logger.info("生产消息：{}", object);
    amqpTemplate.convertAndSend(routingKey, object);
}

4.maven依赖：

<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.4.3</version>
</dependency>
<dependency>
    <groupId>org.springframework.amqp</groupId>
    <artifactId>spring-rabbit</artifactId>
    <version>2.1.4.RELEASE</version>
</dependency>