一、安装

1.docker安装

docker run -d --restart unless-stopped --name rabbitmq  -p 5672:5672 -p 15672:15672 rabbitmq:management

2.docker-compose.yml 安装

version: '3'
services:
  rabbitmq:
    image: rabbitmq:management
    restart: always
    container_name: rabbitmq
    environment:
      - RABBITMQ_DEFAULT_USER=kexianjun
      - RABBITMQ_DEFAULT_PASS=123456
    ports:
      - 15672:15672
      - 5672:5672
    volumes:
      - /usr/local/docker/rabbitmq/data:/var/lib/rabbitmq
    networks:
      - webservice_web-service

networks:
  webservice_web-service:
    external: true

二、导包

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

三、生产方

1.配置文件

spring:
  rabbitmq:
    addresses: ${serverIp}:5672
#    host: ${serverIp}
#    port: 5672
    username: kexianjun
    password: 123456
    virtual-host: /
    connection-timeout: 15000

image.png

2.java代码

@Autowired
private RabbitTemplate template;

public void send（CargoOder cargoOder）{
CorrelationData correlationData=new CorrelationData();
    correlationData.setId(order.getCargoOderId().toString());
    template.convertAndSend("order-exchange","order."+cargoOder.getInsurerId(),cargoOder,correlationData);
}
 

3.采用消息确认模式配置

spring:
  rabbitmq:
    addresses: ${serverIp}:5672
#    host: ${serverIp}
#    port: 5672
    username: kexianjun
    password: 123456
    virtual-host: /
    connection-timeout: 15000
    #采用消息确认模式
    publisher-confirms: true
    publisher-returns: true
    template:
      mandatory: true

4.采用消息确认模式代码

@Component
@Slf4j
public class RabbitOrderSender {
  @Autowired
  private RabbitTemplate rabbitTemplate;
  //回调函数：confirm确认
  private final ConfirmCallback confirmCallback = (correlationData, ack, cause) -> {
    System.err.println("correlationData:" + correlationData);
    Long cargoOrderId = Long.valueOf(correlationData.getId());
    if (ack) {
      //如果confirm返回成功 则进行更新
      System.out.println("rabbitOrderSender:发送数据");
      //TODO 数据库惭怍
    } else {
      //失败则进行具体的后续操作：重试或者补偿等手段
      log.error("===============异常=======================");
      System.err.println("异常处理。。。");
    }
  };

  /**
   * 发送消息方法调用：构建自定义对象消息
   * @param order 需要发送的消息order
   */
  public void sendOrder(CargoOrder order) {
    //监听回调函数
    rabbitTemplate.setConfirmCallback(confirmCallback);
    //消息唯一ID
    CorrelationData correlationData = new CorrelationData(order.getCargoOrderId().toString());

    rabbitTemplate.convertAndSend("order-exchange", "order.ABC", order, correlationData);
  }
}

四、消费方

1.配置文件

spring:
  rabbitmq:
    host: ${serverIp}
    port: 5672
    username: kexianjun
    password: 123456
    listener:
      simple:
        concurrency: 5
        max-concurrency: 10
        prefetch: 1 #并发通行1条
        acknowledge-mode: manual #手动确定消息已经消费，auto是自动
    connection-timeout: 15000

image.png

2.代码

@Component
public class RabbitRecerver {
 @RabbitHandler
  @RabbitListener(bindings = @QueueBinding(value = @Queue(value = "order-queue",durable = "true"),
      exchange =@Exchange(name = "order-exchange",durable = "true",type = "topic"),key = "order.*")
  )
  public void listenerTrade(@Payload CargoOder cargoOder,@Headers Map<String,Object> headers,
      Channel channel) {
    log.info("============收到消息===========：\n{}",cargoOder);
    Long deliveryTay = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
    try {
      //
      channel.basicAck(deliveryTay,false);
    } catch (IOException e) {
      log.error("消息确认失败");
      e.printStackTrace();
    }
  }
}

配置2

spring.rabbitmq.address=192.168.1.102:5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=/
spring.rabbitmq.connection-timeout=15000

spring.rabbitmq.listener.simple.concurrency=5
spring.rabbitmq.listener.simple.max-concurrency=10
spring.rabbitmq.listener.simple.acknowledge-mode=manual
spring.rabbitmq.listener.simple.prefetch=1
spring.rabbitmq.listener.order.queue.name=order-queue
spring.rabbitmq.listener.order.queue.durable=true
spring.rabbitmq.listener.order.exchange.name=order-exchange
spring.rabbitmq.listener.order.exchange.durable=true
spring.rabbitmq.listener.order.exchange.type=topic
spring.rabbitmq.listener.order.exchange.ignoreDeclarationExceptions=true
spring.rabbitmq.listener.order.key=order.*

代码2

@Component
public class RabbitRecerver {
    //从配置文件读取
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "${spring.rabbitmq.listener.order.queue.name}",
            durable = "${spring.rabbitmq.listener.order.queue.durable}"),
            exchange = @Exchange(value = "${spring.rabbitmq.listener.order.exchange.name}",
            durable = "${spring.rabbitmq.listener.order.exchange.durable}",
            type = "${spring.rabbitmq.listener.order.exchange.type}",
            ignoreDeclarationExceptions = "${spring.rabbitmq.listener.order.exchange.ignoreDeclarationExceptions}"),
            key = "${spring.rabbitmq.listener.order.key}"
            )
    )
    @RabbitHandler
    public void onOrderMessage(@Payload Order order, Channel channel, @Headers Map<String,Object> headers) throws IOException {
        System.err.println("------------------------------------------");
        System.out.println("${spring.rabbitmq.listener.order.key}");
        System.out.println("${spring.rabbitmq.listener.order.key}");
        System.out.println("${spring.rabbitmq.listener.order.key}");
        System.err.println("消费端order："+order.getId());
        Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
        
        channel.basicAck(deliveryTag, false);
    }
}