Java Spring Boot 整合RabbitMQ（一）：Hello World -B2B2C小程序電子商務(wù)

Spring Boot 整合

環(huán)境：

RabbitMQ：3.7.4

Spring Boot：2.0.1.RELEASE

因為有 Starter POMs，在 Spring Boot 中整合 RabbitMQ 是一件非常容易的事，其中的 AMQP 模塊就可以很好的支持 RabbitMQ。

我們可以使用 Spring Intializr 或 https://start.spring.io/ 創(chuàng)建一個 Spring Boot 工程，并勾選 RabbitMQ。

或者手動在 pom.xml 文件中加入

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

在 application.yml 中配置關(guān)于 RabbitMQ 的連接和用戶信息，如果沒有改 RabbitMQ 的默認(rèn)配置的話，這里零配置即可啟動。這里我們還定義了一些額外的配置備用。

spring:
  profiles:
    active: usage_message
  rabbitmq:
    port: 5672
tutorial:
  client:
    duration: 10000

生產(chǎn)者

Spring AMQP 讓我們用少量的代碼就能輕松實現(xiàn)消息的發(fā)送和接收。通過注入 AmqpTemplate 接口的實例來實現(xiàn)消息的發(fā)送，AmqpTemplate 接口定義了一套針對 AMQP 協(xié)議的基礎(chǔ)操作。在 Spring Boot 中會根據(jù)配置來注入其具體實現(xiàn) (AmqpTemplate 的默認(rèn)實現(xiàn)就是 RabbitTemplate)。

public class Tut1Sender {
    @Autowired
    private AmqpTemplate template;
    @Autowired
    private Queue queue;
    /**
     * 用定時任務(wù)來模擬生產(chǎn)者定時發(fā)送消息
     */
    @Scheduled (fixedDelay = 1000, initialDelay = 500)
    public void send() {
        String message = "Hello World!" + new Date();
        template.convertAndSend(queue.getName(), message);
        System.out.println(" [x] Sent '" + message + "'");
    }
}

在該生產(chǎn)者中，我們會產(chǎn)生一個字符串，并發(fā)送到名為”hello-world” 的隊列中。

消費者

創(chuàng)建消費者 Receiver。通過 @RabbitListener 注解定義該類對”hello-world” 隊列的監(jiān)聽，并用 @RabbitHandler 注解來指定對消息的處理方法。所以，該消費者實現(xiàn)了對”hello-world” 隊列的消費，消費操作為輸出消息的字符串內(nèi)容。

@RabbitListener(queues = "hello-world")
public class Tut1Receiver {
    @RabbitHandler
    public void receive(String in) {
        System.out.println(" [x] Received '" + in + "'");
    }
}

配置類
創(chuàng)建一個新的 JavaConfig 文件

@Profile({"tut1", "hello-world"})
@Configuration
public class Tut1Config {
    @Bean
    public Queue queue() {
        return new Queue("hello-world");
    }
   @Profile("receiver")
    @Bean
    public Tut1Receiver receiver() {
        return new Tut1Receiver();
    }
   @Profile("sender")
   @Bean 
    public Tut1Sender sender() {
        return new Tut1Sender();
    }
}

在上面的 JavaConfig 中，我們使用 @Configuration 讓 Spring 知道這是一個 Java 配置，并定義了生產(chǎn)者、消費者和一個名為”hello-world” 的隊列。并且，我們使用 Spring Profiles 來控制它運行哪個示例，以及它是生產(chǎn)者還是消費者，這樣我們就可以簡單的通過啟動參數(shù)傳遞我們的配置文件來正確的啟動應(yīng)用了。了解springcloud架構(gòu)可以加求求：三五三六二四七二五九

至于具體的生產(chǎn)者（Tut1Sender）和消費者（Tut1Receiver），我們這里僅先定義出來，稍后再具體實現(xiàn)。

應(yīng)用主類

再小小的改造一下生成的 RabbitmqTutorialApplication.java

@SpringBootApplication
@EnableScheduling
public class RabbitmqTutorialApplication {
    public static void main(String[] args) {
        new SpringApplicationBuilder()
                .sources(RabbitmqTutorialApplication.class)
                // 設(shè)置成非 web 環(huán)境
                .web(WebApplicationType.NONE)
                .run(args);
    }
   @Profile("usage_message")
   @Bean 
    public CommandLineRunner usage() {
        return arg0 -> {
            System.out.println("This app uses Spring Profiles to control its behavior.\n");
            System.out.println("Sample usage: java -jar target/rabbitmq-tutorial-0.0.1-SNAPSHOT.jar --spring.profiles.active=hello-world,sender");
        };
    }
    @Profile("!usage_message")
    @Bean
     public CommandLineRunner tutorial() {
        return new RabbitTutorialRunner();
    }
}

這里我將環(huán)境設(shè)置為了 WebApplicationType.NONE，即非 WEB 環(huán)境，因為默認(rèn)的話 Netty 會監(jiān)聽 8080 端口，同時運行的話就會接口沖突導(dǎo)致啟動失敗（當(dāng)然，也可以直接在啟動時用參數(shù)綁定不同的端口以避免沖突）。

其中的 RabbitTutorialRunner 如下

public class RabbitTutorialRunner implements CommandLineRunner {
   @Value("${tutorial.client.duration:0}")
    private int duration;
    @Autowired
private ConfigurableApplicationContext ctx;
    @Override 
 public void run(String... args) throws Exception {
        System.out.println("Ready ... running for " + duration + "ms");
        Thread.sleep(duration);
        ctx.close();
    }
}

這個 Runner 主要是為了阻止主線程退出。除了用 Thread.sleep(millisecond)，也可以用 CountDownLatch 來達(dá)到相同的目的。

運行

編譯

mvn clean package -Dmaven.test.skip=true

運行

java -jar target/rabbitmq-tutorial-0.0.1-SNAPSHOT.jar --spring.profiles.active=tut1,sender

java -jar target/rabbitmq-tutorial-0.0.1-SNAPSHOT.jar --spring.profiles.active=tut1,receiver

輸出

// Sender

Ready … running for 10000ms

[x] Sent ‘Hello World!Thu Apr 12 16:56:01 CST 2018’

[x] Sent ‘Hello World!Thu Apr 12 16:56:03 CST 2018’

[x] Sent ‘Hello World!Thu Apr 12 16:56:04 CST 2018’

…