SpringBoot：接入RabbitMQ（完整版）

逃离我推掉我的手 2023-02-23 08:40 24阅读 0赞

##### pom #####

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

##### Producer #####

*  application.yml

server:
      port: 8001
      servlet:
        context-path: /
    spring:
      application:
        name: rabbit-producer
      http:
        encoding:
          charset: UTF-8
      jackson:
        date-format: yyyy-MM-dd HH:mm:ss
        default-property-inclusion: NON_NULL
        time-zone: GMT+8
      rabbitmq:
        # 集群地址，用逗号分隔
        addresses: 192.168.11.71:5672,192.168.11.72:5672,192.168.11.71:5673
        connection-timeout: 15000
        password: guest
        # 使用启用消息确认模式
        publisher-confirms: true
        username: guest
        virtual-host: /

*  RabbitSender

import java.util.Map;
    import java.util.UUID;
    
    import org.springframework.amqp.AmqpException;
    import org.springframework.amqp.core.MessagePostProcessor;
    import org.springframework.amqp.rabbit.connection.CorrelationData;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.messaging.Message;
    import org.springframework.messaging.MessageHeaders;
    import org.springframework.messaging.support.MessageBuilder;
    import org.springframework.stereotype.Component;
    
    @Component
    public class RabbitSender { 
    
    	@Autowired
    	private RabbitTemplate rabbitTemplate;
    	
    	/** * 这里就是确认消息的回调监听接口，用于确认消息是否被broker所收到 */
    	final ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() { 
    		/** * @param ack broker 是否落盘成功 * @param cause 失败的一些异常信息 */
    		@Override
    		public void confirm(CorrelationData correlationData, boolean ack, String cause) { 
    			System.err.println("消息ACK结果:" + ack + ", correlationData: " + correlationData.getId());
    		}
    	};
    	
    	/** * 对外发送消息的方法 * @param message 具体的消息内容 * @param properties 额外的附加属性 * @throws Exception */
    	public void send(Object message, Map<String, Object> properties) throws Exception { 
    		
    		MessageHeaders mhs = new MessageHeaders(properties);
    		Message<?> msg = MessageBuilder.createMessage(message, mhs);
    		
    		rabbitTemplate.setConfirmCallback(confirmCallback);
    		
    		// 指定业务唯一的iD
    		CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
    		
    		MessagePostProcessor mpp = new MessagePostProcessor() { 
    			
    			@Override
    			public org.springframework.amqp.core.Message postProcessMessage(org.springframework.amqp.core.Message message)
    					throws AmqpException { 
    				System.err.println("---> post to do: " + message);
    				return message;
    			}
    		};
    		
    		rabbitTemplate.convertAndSend("exchange-1", "springboot.rabbit", msg, mpp, correlationData);
    	
    	}
    }

##### Consumer #####

*  application.yml

server:
      port: 8002
      servlet:
        context-path: /
    spring:
      application:
        name: rabbit-producer
      http:
        encoding:
          charset: UTF-8
      jackson:
        date-format: yyyy-MM-dd HH:mm:ss
        default-property-inclusion: NON_NULL
        time-zone: GMT+8
      rabbitmq:
        addresses: 192.168.11.71:5672,192.168.11.72:5672,192.168.11.71:5673
        connection-timeout: 15000
        listener:
          # 最好不要在代码里写死配置信息，尽量使用这种方式也就是配置文件的方式
          # 在代码里使用 ${} 方式进行设置配置: ${spring.rabbitmq.listener.order.exchange.name}
          order:
            exchange:
              durable: true
              key: order.*
              name: order-exchange
              type: topic
          # 表示消费者消费成功消息以后需要手工的进行签收(ack)，默认为auto
          simple:
            acknowledge-mode: manual
            concurrency: 5
            max-concurrency: 10
            prefetch: 1
        password: guest
        username: guest
        virtual-host: /

*  RabbitReceive

import org.springframework.amqp.rabbit.annotation.Exchange;
    import org.springframework.amqp.rabbit.annotation.Queue;
    import org.springframework.amqp.rabbit.annotation.QueueBinding;
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.amqp.support.AmqpHeaders;
    import org.springframework.messaging.Message;
    import org.springframework.stereotype.Component;
    
    import com.rabbitmq.client.Channel;
    
    @Component
    public class RabbitReceive { 
    	
    	/** * 组合使用监听 * @RabbitListener @QueueBinding @Queue @Exchange * @param message * @param channel * @throws Exception */
    	@RabbitListener(bindings = @QueueBinding(
    					value = @Queue(value = "queue-1", durable = "true"),
    					exchange = @Exchange(name = "exchange-1",
    					durable = "true",
    					type = "topic",
    					ignoreDeclarationExceptions = "true"),
    					key = "springboot.*"
    				)
    			)
    	@RabbitHandler
    	public void onMessage(Message message, Channel channel) throws Exception { 
    		// 1. 收到消息以后进行业务端消费处理
    		System.err.println("-----------------------");
    		System.err.println("消费消息:" + message.getPayload());
    
    		// 2. 处理成功之后 获取deliveryTag 并进行手工的ACK操作, 因为我们配置文件里配置的是 手工签收
    		// spring.rabbitmq.listener.simple.acknowledge-mode=manual
    		Long deliveryTag = (Long)message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
    		channel.basicAck(deliveryTag, false);
    	}
    	
    }