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

#rabbitmqctl add_user 賬號 密碼
rabbitmqctl add_user admin 614
#分配用戶標(biāo)簽(admin為要賦予administrator權(quán)限的剛創(chuàng)建的那個(gè)賬號的名字)
rabbitmqctl set_user_tags admin administrator
#設(shè)置權(quán)限，開啟遠(yuǎn)程訪問
rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*"

server:
 port: 8080
 servlet:
 context-path: /
spring:
 http:
 encoding:
  charset: UTF-8
 jackson:
  #前端頁面?zhèn)鱀ate值時(shí)格式化
  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;//儲存消息發(fā)送的唯一標(biāo)識
 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,//消息體內(nèi)容
        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:
   #前端頁面?zhèn)鱀ate值時(shí)格式化
   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消費(fèi)端配置
  listener:
   simple:
    #并發(fā)數(shù)
    concurrency: 5
    #最大并發(fā)數(shù)
    max-concurrency: 10
    #簽收模式:手工簽收、自動簽收
    acknowledge-mode: manual
    #限流，在此消費(fèi)端同一時(shí)間只有一條消息消費(fèi)
    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表示監(jiān)聽的具體隊(duì)列.
    bindings屬性代表綁定。里邊有幾個(gè)值填寫,填寫好綁定的隊(duì)列名字和交換機(jī)名字
    指定好routingKey。若指定的這些參數(shù)不存在的話。則會自行給你創(chuàng)建好
    durable代表是否持久化
  */
  @RabbitListener(bindings = @QueueBinding(
      value = @Queue(value = "user1-queue", durable = "true"),
      exchange = @Exchange(name = "user1-exchange", durable = "true", type = "topic"),
      key = "user1.#"
  )
  )
  @RabbitHandler//標(biāo)識這個(gè)方法用于消費(fèi)消息
  public void onUser1Message(@Payload User1 user1,
                @Headers Map<String, Object> headers,
                Channel channel) throws IOException {
    //消費(fèi)者操作
    System.out.println("-------收到消息辣！-----");
    System.out.println("發(fā)過來的用戶名為:" + user1.getName());
    //basicAck()表示確認(rèn)已經(jīng)消費(fèi)消息。通知一下mq,需要先得到 delivery tag
    //delivery tag可以從消息頭里邊get出來
    Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
    channel.basicAck(deliveryTag, false);
  }
}