Spring集成rabbitmq消息中间件

一、消息生产者配置

<!-- rabbitmq服务配置-->
<rabbit:connection-factory id="connectionFactory" host="192.168.188.128" username="root" password="root" port="5672" />
<rabbit:admin connection-factory="connectionFactory" />

<!-- 声明队列 -->
<rabbit:queue name="myQueue" id="myQueue" durable="true" auto-delete="false" exclusive="false" />

<!-- 任何发送到Fanout Exchange的消息都会被转发到与该Exchange绑定(Binding)的全部Queue上 -->
<rabbit:direct-exchange name="my_mq_exchange" durable="true" auto-delete="false" id="my_mq_exchange">
   <rabbit:bindings>
      <rabbit:binding queue="myQueue" key="myQueue_key"></rabbit:binding>
   </rabbit:bindings>
</rabbit:direct-exchange>

<rabbit:template id="amqpTemplate" queue="myQueue" routing-key="myQueue_key" connection-factory="connectionFactory" exchange="my_mq_exchange"></rabbit:template>

发送消息

@Autowired
private UserService userService;

@Autowired
private AmqpTemplate amqpTemplate;

@RequestMapping("/test.do")
public void test(){
   userService.test();
   //System.out.println("aaaaaaaaaaaaaaaaaaaaaaaaaaa");
   //System.out.println("使用rabbitmq发送消息啦啦啦啦啦啦啦");
   amqpTemplate.convertAndSend("我是rabbitmq生产者,我在发送消息啦啦啦啦啦啦啦");
}

 

二、消息消费者配置

<!-- 线程池配置 -->
<bean id="taskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
   <!-- 线程池维护线程的最少数量 -->
   <property name="corePoolSize" value="5" />
   <!-- 线程池维护线程所容许的空闲时间 -->
   <property name="keepAliveSeconds" value="30000" />
   <!-- 线程池维护线程的最大数量 -->
   <property name="maxPoolSize" value="1000" />
   <!-- 线程池使用的缓冲队列 -->
   <property name="queueCapacity" value="200" />
</bean>


<!-- rabbitmq服务配置 默认端口为5672-->
<rabbit:connection-factory id="connectionFactory" host="192.168.188.128" username="root" password="root" port="5672" />

<rabbit:admin connection-factory="connectionFactory" />

<!-- 声明队列-->
<rabbit:queue id="myQueue" name="myQueue" durable="true" auto-delete="false" exclusive="false"/>

<!-- 任何发送到Fanout Exchange的消息都会被转发到与该Exchange绑定(Binding)的全部Queue上-->
<rabbit:direct-exchange name="my_mq_exchange" id="my_mq_exchange" durable="true" auto-delete="false">
   <rabbit:bindings>
      <rabbit:binding queue="myQueue" key="myQueue_key"></rabbit:binding>
   </rabbit:bindings>
</rabbit:direct-exchange>

<bean id="MyQueueListener" class="com.mall.listener.rabbitmq.MyQueueListener"></bean>

<!-- queue listener 观察监听模式 当有消息到达时会通知监听在对应的队列上的监听对象-->
<rabbit:listener-container connection-factory="connectionFactory" acknowledge="auto" task-executor="taskExecutor">
   <rabbit:listener queues="myQueue" ref="MyQueueListener"></rabbit:listener>
</rabbit:listener-container>

 

接收消息

public class MyQueueListener implements MessageListener{


    public void onMessage(Message message){

        //System.out.println("我是rabbit消费者,我在消费生产者生产的消息啦啦啦啦啦啦啦");
        System.out.println(new String(message.getBody()));
    }

}