SpringBoot整合RabbitMQ, 实现生产者与消费者的功能

//AMQP

compile( 'org.springframework.boot:spring-boot-starter-amqp:2.0.4.RELEASE' )

#rabbitmqctl add_user 账号 密码

rabbitmqctl add_user admin 614

#分配用户标签(admin为要赋予administrator权限的刚创建的那个账号的名字)

rabbitmqctl set_user_tags admin administrator

#设置权限，开启远程访问

rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*"

server:

  port: 8080

  servlet:

  context-path: /

spring:

  http:

  encoding:

   charset: UTF-8

  jackson:

   #前端页面传Date值时格式化

   date - format : yyyy-MM- dd HH:mm:ss

   time -zone: GMT+8

  datasource:

   driver-class-name: com.mysql.cj.jdbc.Driver

   url: jdbc:mysql: //192 .168.194.128:3306 /mysql ?serverTimezone=Asia /Shanghai

   username: root

   password: 614

  rabbitmq:

  port: 5672

  host: 192.168.194.128

  username: admin

  password: 614

  virtual-host: /

  connection-timeout: 15s

#Redis配置

  redis:

  host: 192.168.194.128

  port: 6379

  #Redis连接池配置

  jedis:

   pool:

   min-idle: 0

   max-idle: 8

   max-active: 8

   max-wait: -1ms

package com.skypyb.rabbitmq.entity;

import java.io.Serializable;

public class User1 implements Serializable{

  private Long id;

  private String name;

  private String messageId; //储存消息发送的唯一标识

  public User1() {

  }

  public User1(Long id, String name, String messageId) {

   this .id = id;

   this .name = name;

   this .messageId = messageId;

  }

  public Long getId() {

   return id;

  }

  public void setId(Long id) {

   this .id = id;

  }

  public String getName() {

   return name;

  }

  public void setName(String name) {

   this .name = name;

  }

  public String getMessageId() {

   return messageId;

  }

  public void setMessageId(String messageId) {

   this .messageId = messageId;

  }

}

package com.skypyb.rabbitmq.producer;

import com.skypyb.rabbitmq.entity.User1;

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 ( "user1Sender" )

public class User1Sender {

   @Autowired

   private RabbitTemplate rabbitTemplate; //操作rabbitmq的模板

   public void send(User1 user1){

     CorrelationData correlationData= new CorrelationData();

     correlationData.setId(user1.getMessageId());

     rabbitTemplate.convertSendAndReceive(

         "user1-exchange" , //exchange

         "user1.key1" , //routingKey

         user1, //消息体内容

         correlationData //消息唯一ID

     );

   }

}

package com.skypyb.test;

import com.skypyb.rabbitmq.Application;

import com.skypyb.rabbitmq.entity.User1;

import com.skypyb.rabbitmq.producer.User1Sender;

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 javax.annotation.Resource;

import java.util.UUID;

@RunWith (SpringRunner. class )

@SpringBootTest (classes = Application. class )

public class TestOne {

   @Autowired

   private User1Sender user1Sender;

   @Test

   public void testSend1(){

     User1 user1 = new User1();

     user1.setId(1L);

     user1.setName( "测试用户1" );

     user1.setMessageId( "user1$" +System.currentTimeMillis()+ "$" + UUID.randomUUID().toString());

     user1Sender.send(user1);

   }

}

server:

  port: 8081

  servlet:

   context-path: /

spring:

  http:

   encoding:

    charset: UTF-8

  jackson:

    #前端页面传Date值时格式化

    date - format : yyyy-MM- dd HH:mm:ss

    time -zone: GMT+8

  datasource:

    driver-class-name: com.mysql.cj.jdbc.Driver

    url: jdbc:mysql: //192 .168.194.128:3306 /mysql ?serverTimezone=Asia /Shanghai

    username: root

    password: 614

  #rabbitmq基本配置

  rabbitmq:

   addresses: 192.168.194.128:5672

   username: admin

   password: 614

   virtual-host: /

   connection-timeout: 15s

  #rabbitmq消费端配置

   listener:

    simple:

     #并发数

     concurrency: 5

     #最大并发数

     max-concurrency: 10

     #签收模式:手工签收、自动签收

     acknowledge-mode: manual

     #限流，在此消费端同一时间只有一条消息消费

     prefetch: 1

#Redis配置

  redis:

   host: 192.168.194.128

   port: 6379

   #Redis连接池配置

   jedis:

    pool:

     min-idle: 0

     max-idle: 8

     max-active: 8

     max-wait: -1ms

package com.skypyb.rabbitmq.controller;

import com.rabbitmq.client.Channel;

import com.skypyb.rabbitmq.entity.User1;

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

import org.springframework.amqp.support.AmqpHeaders;

import org.springframework.messaging.handler.annotation.Headers;

import org.springframework.messaging.handler.annotation.Payload;

import org.springframework.stereotype.Component;

import java.io.IOException;

import java.util.Map;

@Component

public class User1Receiver {

   /**

    * @param user1  消息体,使用 @Payload 注解

    * @param headers 消息头,使用 @Headers 注解

    * @param channel

    */

   /*@RabbitListener表示监听的具体队列.

     bindings属性代表绑定。里边有几个值填写,填写好绑定的队列名字和交换机名字

     指定好routingKey。若指定的这些参数不存在的话。则会自行给你创建好

     durable代表是否持久化

   */

   @RabbitListener (bindings = @QueueBinding (

       value = @Queue (value = "user1-queue" , durable = "true" ),

       exchange = @Exchange (name = "user1-exchange" , durable = "true" , type = "topic" ),

       key = "user1.#"

   )

   )

   @RabbitHandler //标识这个方法用于消费消息

   public void onUser1Message( @Payload User1 user1,

                 @Headers Map<String, Object> headers,

                 Channel channel) throws IOException {

     //消费者操作

     System.out.println( "-------收到消息辣！-----" );

     System.out.println( "发过来的用户名为:" + user1.getName());

     //basicAck()表示确认已经消费消息。通知一下mq,需要先得到 delivery tag

     //delivery tag可以从消息头里边get出来

     Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);

     channel.basicAck(deliveryTag, false );

   }

}