SpringBoot整合RabbitMQ, 实现生产者与消费者的功能 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);
  }
}