rabbitmq系统学习（二）

2023-08-14 08:47:46

Rabbitmq高级整合应用

RabbitMq整合Spring AMQP实战

RabbitAdmin
- 使用RabbitTemplate的execute方法执行对应操作
- rabbitAdmin.declareExchange()//声明
- rabbitAdmin.declareQueue()
- rabbitAdmin.declareBinding()
- rabbitAdmin.declareBinding(BindingBuilder.bind(new Queue("test.topic.queue",false)).//直接创建队列
  
  to(new FanoutExchange("test.topic",false,false)).//直接创建交换机，建立关联关系
  
  with("user.#"));//指定路由Key
- rabbitAdmin.declareBinding(BindingBuilder.bind(new Queue("test.fanout.queue",false)).to(new FanoutExchange("test.fanout",false,false)));
- rabbitAdmin.purgeQueue("test.topic.queue",false);//清空队列数据
- autoStartup必须设置为true，否则Spring容器不会加载RabbitAdmin类
- RabbitAdmin底层实现就是从Spring容器中获取Exchange、Binding、RoutingKey以及Queue的@Bean声明
消息模板-RabbitTemplate
- 发送消息的关键类
- 提供了丰富的发送消息方法，包括可靠性投递消息方法、回调监听消息接口ConfirmCallback、返回值确认接口ReturnCallback等等。同样我们需要进行注入到Spring容器中，然后直接使用
SimpleMessageListenerContainer
- 简单消息监听器，可以监听队列（多个队列），自动启动、自动声明功能
- 设置事务特性、事务管理器、事务属性、事务容量（并发）、是否开启事务、回滚消息等
- 设置消费者数量、最小最大数量、批量消费
- 设置消息确认和自动确认模式、是否重回队列、异常捕获handler函数
- 设置消费者标签生成策略、是否独占模式、消费者属性等
- 设置具体的监听器、消息转换器等等
- 注意：SimpleMessageListenerContainer可以进行动态设置，比如在运行中的应用可以动态的修改其消费者数量的大小、接收消息的模式等
- 很多基于RabbitMQ的自动化后端管控台在进行动态设置的时候，也是根据这一特性去实现的。
MessageListenerAdapter
- 消息监听适配器
- 可以自定义类来接收处理消息
- 队列和方法名称一一匹配
  - adapter.setQueueOrTagToMethodName(Map)
MessageConverter
- 可以用来转换消息，比如把byte[] 转成String之类
- 要用Map接，用json格式的转换器，Jackson2JsonMessageConverter
- 支持java对象转换，用DefaultJackson2JavaTypeMapper，识别header中__TypeId__，值为类的全路径
- 支持java对象多映射关系
- ext convert 全局转换器ContentTypeDelegatingMessageConverter
文件操作
- byte[] body = Files.readAllBytes(Paths.get("d:/002_books","picture.png"));

4-12

RabbitMQ整合SpringBoot实战

publish-confirms,实现一个监听器用于监听Broker端给我们返回的确认请求RabbitTemplate.ConfirmCallback
注意一点，在发送消息的时候对template进行配置mandatory=true保证监听有效
生产端还可以配置其他属性，比如发送重试，超时时间，次数，间隔等
代码，服务端

package com.bfxy.springboot.conusmer;

import java.util.Map;

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.messaging.handler.annotation.Headers;

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

import org.springframework.stereotype.Component;

import com.rabbitmq.client.Channel;

@Component

public class RabbitReceiver {

	@RabbitListener(bindings = @QueueBinding(

			value = @Queue(value = "queue-1",

			durable="true"),

			exchange = @Exchange(value = "exchange-1",

			durable="true",

			type= "topic",

			ignoreDeclarationExceptions = "true"),

			key = "springboot.*"

			)

	)

	@RabbitHandler

	public void onMessage(Message message, Channel channel) throws Exception {

		System.err.println("--------------------------------------");

		System.err.println("消费端Payload: " + message.getPayload());

		Long deliveryTag = (Long)message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);

		//手工ACK

		channel.basicAck(deliveryTag, false);

	}

	/**

	 *

	 * 	spring.rabbitmq.listener.order.queue.name=queue-2

		spring.rabbitmq.listener.order.queue.durable=true

		spring.rabbitmq.listener.order.exchange.name=exchange-1

		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=springboot.*

	 * @param order

	 * @param channel

	 * @param headers

	 * @throws Exception

	 */

	@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 com.bfxy.springboot.entity.Order order,

			Channel channel,

			@Headers Map<String, Object> headers) throws Exception {

		System.err.println("--------------------------------------");

		System.err.println("消费端order: " + order.getId());

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

		//手工ACK

		channel.basicAck(deliveryTag, false);

	}

}

- 配置

spring.rabbitmq.addresses=i.tzxylao.com:5672,i.tzxylao.com:5673,i.tzxylao.com:5674

spring.rabbitmq.username=dev

spring.rabbitmq.password=xxxx

spring.rabbitmq.virtual-host=hospital

spring.rabbitmq.connection-timeout=15000

spring.rabbitmq.listener.simple.acknowledge-mode=manual

spring.rabbitmq.listener.simple.concurrency=5

spring.rabbitmq.listener.simple.max-concurrency=10

spring.rabbitmq.listener.order.queue.name=queue-2

spring.rabbitmq.listener.order.queue.durable=true

spring.rabbitmq.listener.order.exchange.name=exchange-2

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=springboot.*

代码，生产端

package com.bfxy.springboot.producer;

import java.util.Map;

import org.springframework.amqp.rabbit.core.RabbitTemplate;

import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;

import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;

import org.springframework.amqp.rabbit.support.CorrelationData;

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;

import com.bfxy.springboot.entity.Order;

@Component

public class RabbitSender {

	//自动注入RabbitTemplate模板类

	@Autowired

	private RabbitTemplate rabbitTemplate;  

	//回调函数: confirm确认

	final ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {

		@Override

		public void confirm(CorrelationData correlationData, boolean ack, String cause) {

			System.err.println("correlationData: " + correlationData);

			System.err.println("ack: " + ack);

			if(!ack){

				System.err.println("异常处理....");

			}

		}

	};

	//回调函数: return返回

	final ReturnCallback returnCallback = new RabbitTemplate.ReturnCallback() {

		@Override

		public void returnedMessage(org.springframework.amqp.core.Message message, int replyCode, String replyText,

				String exchange, String routingKey) {

			System.err.println("return exchange: " + exchange + ", routingKey: "

				+ routingKey + ", replyCode: " + replyCode + ", replyText: " + replyText);

		}

	};

	//发送消息方法调用: 构建Message消息

	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);

		rabbitTemplate.setReturnCallback(returnCallback);

		//id + 时间戳 全局唯一

		CorrelationData correlationData = new CorrelationData("1234567890");

		rabbitTemplate.convertAndSend("exchange-1", "springboot.abc", msg, correlationData);

	}

	//发送消息方法调用: 构建自定义对象消息

	public void sendOrder(Order order) throws Exception {

		rabbitTemplate.setConfirmCallback(confirmCallback);

		rabbitTemplate.setReturnCallback(returnCallback);

		//id + 时间戳 全局唯一

		CorrelationData correlationData = new CorrelationData("0987654321");

		rabbitTemplate.convertAndSend("exchange-2", "springboot.def", order, correlationData);

	}

}

- 配置

spring.rabbitmq.addresses=192.168.11.76:5672

spring.rabbitmq.username=dev

spring.rabbitmq.password=123

spring.rabbitmq.virtual-host=hospital

spring.rabbitmq.connection-timeout=15000

spring.rabbitmq.publisher-confirms=true

spring.rabbitmq.publisher-returns=true

spring.rabbitmq.template.mandatory=true

RabbitMQ整合Spring Cloud实战

Spring Cloud Stream整合

Barista接口：Barista接口是定义来作为后面类的参数，这一接口定义来通道类型和通道名称，通道名称是作为配置用，通道类型则决定了app会使用这一通道进行发送消息还是从中接收消息
@Output：输出注解，用于定义发送消息接口
@Input：输入注解，用于定义消息的消费者接口
@StreamListener：用于定义监听方法的注解
添加代理层，用于快速的替换mq
代码

生产者

package com.bfxy.rabbitmq.stream;

import org.springframework.cloud.stream.annotation.Output;

import org.springframework.messaging.MessageChannel;

/**

 * <B>中文类名：</B><BR>

 * <B>概要说明：</B><BR>

 * 这里的Barista接口是定义来作为后面类的参数，这一接口定义来通道类型和通道名称。

 * 通道名称是作为配置用，通道类型则决定了app会使用这一通道进行发送消息还是从中接收消息。

 */

public interface Barista {

    //String INPUT_CHANNEL = "input_channel";

    String OUTPUT_CHANNEL = "output_channel";  

    //注解@Input声明了它是一个输入类型的通道，名字是Barista.INPUT_CHANNEL，也就是position3的input_channel。这一名字与上述配置app2的配置文件中position1应该一致，表明注入了一个名字叫做input_channel的通道，它的类型是input，订阅的主题是position2处声明的mydest这个主题

//    @Input(Barista.INPUT_CHANNEL)

//    SubscribableChannel loginput();

    //注解@Output声明了它是一个输出类型的通道，名字是output_channel。这一名字与app1中通道名一致，表明注入了一个名字为output_channel的通道，类型是output，发布的主题名为mydest。

    @Output(Barista.OUTPUT_CHANNEL)

    MessageChannel logoutput();  

//	String INPUT_BASE = "queue-1";

//	String OUTPUT_BASE = "queue-1";

//	@Input(Barista.INPUT_BASE)

//	SubscribableChannel input1();

//	MessageChannel output1();  

}

package com.bfxy.rabbitmq.stream;

import java.util.Map;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.cloud.stream.annotation.EnableBinding;

import org.springframework.messaging.Message;

import org.springframework.messaging.MessageHeaders;

import org.springframework.messaging.support.MessageBuilder;

import org.springframework.stereotype.Service;

@EnableBinding(Barista.class)

@Service

public class RabbitmqSender {  

    @Autowired

    private Barista barista;  

    // 发送消息

    public String sendMessage(Object message, Map<String, Object> properties) throws Exception {

        try{

        	MessageHeaders mhs = new MessageHeaders(properties);

        	Message msg = MessageBuilder.createMessage(message, mhs);

            boolean sendStatus = barista.logoutput().send(msg);

            System.err.println("--------------sending -------------------");

            System.out.println("发送数据：" + message + ",sendStatus: " + sendStatus);

        }catch (Exception e){

        	System.err.println("-------------error-------------");

        	e.printStackTrace();

            throw new RuntimeException(e.getMessage());

        }

        return null;

    }  

}

配置文件

server.port=8001

server.servlet.context-path=/producer

spring.application.name=producer

spring.cloud.stream.bindings.output_channel.destination=exchange-3

spring.cloud.stream.bindings.output_channel.group=queue-3

spring.cloud.stream.bindings.output_channel.binder=rabbit_cluster

spring.cloud.stream.binders.rabbit_cluster.type=rabbit

spring.cloud.stream.binders.rabbit_cluster.environment.spring.rabbitmq.addresses=192.168.11.76:5672

spring.cloud.stream.binders.rabbit_cluster.environment.spring.rabbitmq.username=guest

spring.cloud.stream.binders.rabbit_cluster.environment.spring.rabbitmq.password=guest

spring.cloud.stream.binders.rabbit_cluster.environment.spring.rabbitmq.virtual-host=/

消费者

package com.bfxy.rabbitmq.stream;

import org.springframework.cloud.stream.annotation.Input;

import org.springframework.messaging.SubscribableChannel;

/**

 * <B>中文类名：</B><BR>

 * <B>概要说明：</B><BR>

 * 这里的Barista接口是定义来作为后面类的参数，这一接口定义来通道类型和通道名称。

 * 通道名称是作为配置用，通道类型则决定了app会使用这一通道进行发送消息还是从中接收消息。

 * @author ashen（Alienware）

 * @since 2016年7月22日

 */

public interface Barista {

    String INPUT_CHANNEL = "input_channel";  

    //注解@Input声明了它是一个输入类型的通道，名字是Barista.INPUT_CHANNEL，也就是position3的input_channel。这一名字与上述配置app2的配置文件中position1应该一致，表明注入了一个名字叫做input_channel的通道，它的类型是input，订阅的主题是position2处声明的mydest这个主题

    @Input(Barista.INPUT_CHANNEL)

    SubscribableChannel loginput();  

}

package com.bfxy.rabbitmq.stream;

import org.springframework.amqp.support.AmqpHeaders;

import org.springframework.cloud.stream.annotation.EnableBinding;

import org.springframework.cloud.stream.annotation.StreamListener;

import org.springframework.messaging.Message;

import org.springframework.stereotype.Service;

import com.rabbitmq.client.Channel;

@EnableBinding(Barista.class)

@Service

public class RabbitmqReceiver {  

    @StreamListener(Barista.INPUT_CHANNEL)

    public void receiver(Message message) throws Exception {

		Channel channel = (com.rabbitmq.client.Channel) message.getHeaders().get(AmqpHeaders.CHANNEL);

		Long deliveryTag = (Long) message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);

    	System.out.println("Input Stream 1 接受数据：" + message);

    	System.out.println("消费完毕------------");

    	channel.basicAck(deliveryTag, false);

    }

}

配置文件

server.port=8002

server.context-path=/consumer

spring.application.name=consumer

spring.cloud.stream.bindings.input_channel.destination=exchange-3

spring.cloud.stream.bindings.input_channel.group=queue-3

spring.cloud.stream.bindings.input_channel.binder=rabbit_cluster

spring.cloud.stream.bindings.input_channel.consumer.concurrency=1

spring.cloud.stream.rabbit.bindings.input_channel.consumer.requeue-rejected=false

spring.cloud.stream.rabbit.bindings.input_channel.consumer.acknowledge-mode=MANUAL

spring.cloud.stream.rabbit.bindings.input_channel.consumer.recovery-interval=3000

spring.cloud.stream.rabbit.bindings.input_channel.consumer.durable-subscription=true

spring.cloud.stream.rabbit.bindings.input_channel.consumer.max-concurrency=5

spring.cloud.stream.binders.rabbit_cluster.type=rabbit

spring.cloud.stream.binders.rabbit_cluster.environment.spring.rabbitmq.addresses=192.168.11.76:5672

spring.cloud.stream.binders.rabbit_cluster.environment.spring.rabbitmq.username=guest

spring.cloud.stream.binders.rabbit_cluster.environment.spring.rabbitmq.password=guest

spring.cloud.stream.binders.rabbit_cluster.environment.spring.rabbitmq.virtual-host=/

码农公寓

Rabbitmq高级整合应用

RabbitMq整合Spring AMQP实战

RabbitMQ整合SpringBoot实战

RabbitMQ整合Spring Cloud实战

Spring Cloud Stream整合

相关文章