Spring Boot系列教程之7步集成RabbitMQ的方法

<!--rabbitmq-->

<dependency>

  <groupid>org.springframework.boot</groupid>

  <artifactid>spring-boot-starter-amqp</artifactid>

</dependency>

spring: rabbitmq:

  host: 192.168 . 1.161

  port: 5672

  username: guest

  password: guest

  cache:

  channel: size: 10

  listener:

  type: simple

  simple:

  acknowledge-mode: auto

  concurrency: 5

  default -requeue-rejected: true

  max-concurrency: 100

  retry:

  enabled: true # initial-interval: 1000ms

  max-attempts: 3 # max-interval: 1000ms

  multiplier: 1

  stateless: true # publisher-confirms: true </pre>

package com.jackie.springbootdemo.config;

import org.springframework.amqp.core.*;

import org.springframework.amqp.rabbit.config.simplerabbitlistenercontainerfactory;

import org.springframework.amqp.rabbit.connection.connectionfactory;

import org.springframework.amqp.rabbit.core.rabbittemplate;

import org.springframework.amqp.support.converter.jackson2jsonmessageconverter;

import org.springframework.beans.factory.initializingbean;

import org.springframework.beans.factory.annotation.autowired;

import org.springframework.beans.factory.annotation.qualifier;

import org.springframework.beans.factory.annotation.value;

import org.springframework.beans.factory.config.configurablebeanfactory;

import org.springframework.context.annotation.bean;

import org.springframework.context.annotation.configuration;

import org.springframework.context.annotation.scope;

import java.util.hashmap;

import java.util.map;

@configuration public class rabbitmqconfig implements initializingbean { @autowired

  simplerabbitlistenercontainerfactory simplerabbitlistenercontainerfactory;

  @override

  public void afterpropertiesset() throws exception {

  simplerabbitlistenercontainerfactory.setmessageconverter( new jackson2jsonmessageconverter());

  } @bean ( "jackson2jsonmessageconverter" )

  public jackson2jsonmessageconverter jackson2jsonmessageconverter(connectionfactory connectionfactory) {

  return new jackson2jsonmessageconverter();

  } @bean ( "rabbittemplate" )

  @scope (configurablebeanfactory.scope_singleton)

  public rabbittemplate rabbittemplate(connectionfactory connectionfactory,

   @qualifier ( "jackson2jsonmessageconverter" ) jackson2jsonmessageconverter jackson2jsonmessageconverter) {

  rabbittemplate template = new rabbittemplate(connectionfactory);

  template.setmessageconverter( new jackson2jsonmessageconverter());

  return template;

  } // --------------------- 声明队列 ------------------------

  @bean

  public queue demoqueue() {

  return new queue( "demo_queue" );

  } // --------------------- 声明exchange ------------------------ @bean

  public directexchange demoexchange() {

  return new directexchange( "demo_exchange" );

  } // --------------------- 队列绑定 ------------------------

  @bean

  public binding bindingalbumitemcreatedqueue(directexchange demoexchange,

   queue demoqueue) {

  return bindingbuilder.bind(demoqueue).to(demoexchange).with( "100" );

  } }

package com.jackie.springbootdemo.message;

import org.springframework.amqp.rabbit.core.rabbittemplate;

import org.springframework.amqp.rabbit.support.correlationdata;

import org.springframework.beans.factory.annotation.autowired;

import org.springframework.stereotype.component;

@component public class sender implements rabbittemplate.confirmcallback { private rabbittemplate rabbittemplate;

  /**

  * 构造方法注入 */ @autowired

  public sender(rabbittemplate rabbittemplate) {

  this .rabbittemplate = rabbittemplate;

  rabbittemplate.setconfirmcallback( this ); //rabbittemplate如果为单例的话，那回调就是最后设置的内容

  } public void sendmsg(string content) {

  rabbittemplate.convertandsend( "demo_exchange" , "100" , content);

  } /**

  * 回调 */ @override

  public void confirm(correlationdata correlationdata, boolean ack, string cause) {

  system.out.println( " 回调id:" + correlationdata);

  if (ack) {

  system.out.println( "消息成功消费" );

  } else {

  system.out.println( "消息消费失败:" + cause);

  }

  } }

package com.jackie.springbootdemo;

import com.jackie.springbootdemo.message.sender;

import org.junit.test;

import org.junit.runner.runwith;

import org.springframework.beans.factory.annotation.autowired;

import org.springframework.boot.test.context.springboottest;

import org.springframework.test.context.junit4.springrunner;

import org.springframework.test.context.web.webappconfiguration;

@runwith (springrunner. class ) @springboottest (classes = springbootdemoapplication. class ) @webappconfiguration public class rabbitapplicationtests { @autowired

  sender sender;

  @test

  public void contextloads() throws exception {

  sender.sendmsg( "test" );

  } }

package com.jackie.springbootdemo.message;

import org.springframework.amqp.rabbit.annotation.rabbitlistener;

import org.springframework.stereotype.component;

@component public class receiver { @rabbitlistener (queues = "demo_queue" )

  public void created(string message) {

  system.out.println( "orignal message: " + message);

  } }

@rabbitlistener (queues = "demo_queue" ) public class receiver { @rabbithandler public void processmessage1(string message) {

  system.out.println(message);

  } @rabbithandler

  public void processmessage2( byte [] message) {

  system.out.println( new string(message));

  } }