RabbitMQ--整合SpringAMQP（笔记）

Bertha 。 2023-07-11 12:53 1阅读 0赞

### 目录： ###

*  RabbitAdmin
 *  SpringAMQP
 *  RabbitTemplate
 *  SimpleMessageListenerContainer
 *  MessageListenerAdapter
 *  MessageConverter

**pom**

<dependency>
    			<groupId>com.rabbitmq</groupId>
    			<artifactId>amqp-client</artifactId>
    			<version>3.6.5</version>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework.boot</groupId>
    			<artifactId>spring-boot-starter-amqp</artifactId>
    		</dependency>

配置RabbitAdmin

setAutoStartup 必须设置成true 否则Spring容器不会加载RabbitAdmin

@Configuration
    public class RabbitMQConfig {
    
    	@Bean
    	public ConnectionFactory connectionFactory(){
    		CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
    		connectionFactory.setAddresses("192.168.11.76:5672");
    		connectionFactory.setUsername("guest");
    		connectionFactory.setPassword("guest");
    		connectionFactory.setVirtualHost("/");
    		return connectionFactory;
    	}
    	
    	@Bean
    	public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
    		RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
    		rabbitAdmin.setAutoStartup(true);
    		return rabbitAdmin;
    	}
    }

设置交换机、队列、绑定关系

@Autowired
    	private RabbitAdmin rabbitAdmin;
    	
    	@Test
    	public void testAdmin() throws Exception {
                    //exchange
    		rabbitAdmin.declareExchange(new DirectExchange("test.direct", false, false));
    		rabbitAdmin.declareExchange(new TopicExchange("test.topic", false, false));
    		rabbitAdmin.declareExchange(new FanoutExchange("test.fanout", false, false));
    		//queue
    		rabbitAdmin.declareQueue(new Queue("test.direct.queue", false));
    		rabbitAdmin.declareQueue(new Queue("test.topic.queue", false));
    		rabbitAdmin.declareQueue(new Queue("test.fanout.queue", false));
    		//Binding
    		rabbitAdmin.declareBinding(new Binding("test.direct.queue",
    				Binding.DestinationType.QUEUE,
    				"test.direct", "direct", new HashMap<>()));
    
    	}

直接在绑定中创建

rabbitAdmin.declareBinding(
    				BindingBuilder
    				.bind(new Queue("test.topic.queue", false))		//直接创建队列
    				.to(new TopicExchange("test.topic", false, false))	//直接创建交换机 建立关联关系
    				.with("user.#"));	//指定路由Key
    		
    		rabbitAdmin.declareBinding(
    				BindingBuilder
    				.bind(new Queue("test.fanout.queue", false))		
    				.to(new FanoutExchange("test.fanout", false, false)));

## SpringAMQP ##

创建交换机，队列，并绑定关系

/**  
         * 针对消费者配置  
         * 1. 设置交换机类型  
         * 2. 将队列绑定到交换机  
            FanoutExchange: 将消息分发到所有的绑定队列，无routingkey的概念  
            HeadersExchange ：通过添加属性key-value匹配  
            DirectExchange:按照routingkey分发到指定队列  
            TopicExchange:多关键字匹配  
         */  
        @Bean  
        public TopicExchange exchange001() {  
            return new TopicExchange("topic001", true, false);  
        }  
    
        @Bean  
        public Queue queue001() {  
            return new Queue("queue001", true); //队列持久  
        }  
        
        @Bean  
        public Binding binding001() {  
            return BindingBuilder.bind(queue001()).to(exchange001()).with("spring.*");  
        }  
        
        @Bean  
        public TopicExchange exchange002() {  
            return new TopicExchange("topic002", true, false);  
        }  
        
        @Bean  
        public Queue queue002() {  
            return new Queue("queue002", true); //队列持久  
        }
        
        @Bean  
        public Binding binding002() {  
            return BindingBuilder.bind(queue002()).to(exchange002()).with("rabbit.*");  
        } 
        
        @Bean  
        public Queue queue003() {  
            return new Queue("queue003", true); //队列持久  
        }
        
        @Bean  
        public Binding binding003() {  
            return BindingBuilder.bind(queue003()).to(exchange001()).with("mq.*");  
        } 
        
        @Bean  
        public Queue queue_image() {  
            return new Queue("image_queue", true); //队列持久  
        }
        
        @Bean  
        public Queue queue_pdf() {  
            return new Queue("pdf_queue", true); //队列持久  
        }

## RabbitTemplate ##

Producer

@Bean
        public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        	RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        	return rabbitTemplate;
        }

这里是给消息添加额往外包装

@Test
    	public void testSendMessage() throws Exception {
    		//1 创建消息
    		MessageProperties messageProperties = new MessageProperties();
    		messageProperties.getHeaders().put("desc", "信息描述..");
    		messageProperties.getHeaders().put("type", "自定义消息类型..");
    		Message message = new Message("Hello RabbitMQ".getBytes(), messageProperties);
    		
    		rabbitTemplate.convertAndSend("topic001", "spring.amqp", message, new MessagePostProcessor() {
                
    			@Override
    			public Message postProcessMessage(Message message) throws AmqpException {
    				System.err.println("------添加额外的设置---------");
    				message.getMessageProperties().getHeaders().put("desc", "额外修改的信息描述");
    				message.getMessageProperties().getHeaders().put("attr", "额外新加的属性");
    				return message;
    			}
    		});
    	}

## SimpleMessageListenerContainer(简单消息监听器) ##

*  监听多个队列、自动启动、自动声明功能
 *  设置事务特性、事务管理器、事务属性、事务容量（并发）、开启事务、回滚消息等
 *  设置消费者数量、最小最大数量、批量消费
 *  设置消息确认和自动确认模式、是否重回队列、异常捕获handler函数
 *  设置消费者标签生成策略、是否独占模式、消费者属性等
 *  设置具体的监听器、消息转换器等
 *  可以动态设置，比如在运行中可以动态修改消费者数量大小，消息接收模式等。

就是绑定消费策略，然后绑定消费者

**MessageListenerAdapter & MessageConverter**

定义消费方法类

public class MessageDelegate {
    
    	public void handleMessage(byte[] messageBody) {
    		System.err.println("默认方法, 消息内容:" + new String(messageBody));
    	}
    	
    	public void consumeMessage(byte[] messageBody) {
    		System.err.println("字节数组方法, 消息内容:" + new String(messageBody));
    	}
    	
    	public void consumeMessage(String messageBody) {
    		System.err.println("字符串方法, 消息内容:" + messageBody);
    	}
    	
    	public void method1(String messageBody) {
    		System.err.println("method1 收到消息内容:" + new String(messageBody));
    	}
    	
    	public void method2(String messageBody) {
    		System.err.println("method2 收到消息内容:" + new String(messageBody));
    	}
    	
    	
    	public void consumeMessage(Map messageBody) {
    		System.err.println("map方法, 消息内容:" + messageBody);
    	}
    	
    	
    	public void consumeMessage(Order order) {
    		System.err.println("order对象, 消息内容, id: " + order.getId() + 
    				", name: " + order.getName() + 
    				", content: "+ order.getContent());
    	}
    	
    	public void consumeMessage(Packaged pack) {
    		System.err.println("package对象, 消息内容, id: " + pack.getId() + 
    				", name: " + pack.getName() + 
    				", content: "+ pack.getDescription());
    	}
    	
    	public void consumeMessage(File file) {
    		System.err.println("文件对象 方法, 消息内容:" + file.getName());
    	}
    }

### 1、这里会使用上面的consumeMessage 方法消费 （如果不设置setDefaultListenerMethod）会使用默认的handleMessage 消费 ###

这里使用了TextMessageConverter  所有会是consumeMessage 方法的String类型消费（不指定是byte\[\] 类型接收）

转换器 实现，下面会单独列举

@Bean
        public SimpleMessageListenerContainer messageContainer(ConnectionFactory connectionFactory) {
        	SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        	container.setQueues(queue001(), queue002(), queue003(), queue_image(), queue_pdf());
        	container.setConcurrentConsumers(1);
        	container.setMaxConcurrentConsumers(5);
        	container.setDefaultRequeueRejected(false);//是否重回队列
        	container.setAcknowledgeMode(AcknowledgeMode.AUTO);//自动签收
        	container.setExposeListenerChannel(true);//队列监听打开
            //设置消息标签
        	container.setConsumerTagStrategy(new ConsumerTagStrategy() {
    			@Override
    			public String createConsumerTag(String queue) {
    				return queue + "_" + UUID.randomUUID().toString();
    			}
    		});
        	//适配器方式. 默认是有自己的方法名字的：handleMessage
        		// 可以自己指定一个方法的名字: consumeMessage
        		// 也可以添加一个转换器: 从字节数组转换为String
        	MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());
        	adapter.setDefaultListenerMethod("consumeMessage");
        	adapter.setMessageConverter(new TextMessageConverter());
        	container.setMessageListener(adapter);
        	return container;
        }

**对应的Producer**

@Test
    	public void testSendMessage2() throws Exception {
    		//1 创建消息
    		MessageProperties messageProperties = new MessageProperties();
    		messageProperties.setContentType("text/plain");
    		Message message = new Message("mq 消息1234".getBytes(), messageProperties);
    		
    		rabbitTemplate.send("topic001", "spring.abc", message);
    		rabbitTemplate.convertAndSend("topic001", "spring.amqp", "hello object message send!");
    		rabbitTemplate.convertAndSend("topic002", "rabbit.abc", "hello object message send!");
    	}

2、我们的队列名称 和 方法名称 也可以进行一一的匹配

替换上面的适配器方法，queue001队列的消息 method1方法 消费。2和2消费。

MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());
        	adapter.setMessageConverter(new TextMessageConverter());
        	Map<String, String> queueOrTagToMethodName = new HashMap<>();
        	queueOrTagToMethodName.put("queue001", "method1");
        	queueOrTagToMethodName.put("queue002", "method2");
        	adapter.setQueueOrTagToMethodName(queueOrTagToMethodName);
        	container.setMessageListener(adapter);

**对应的Producer**

@Test
    	public void testSendMessage4Text() throws Exception {
    		//1 创建消息
    		MessageProperties messageProperties = new MessageProperties();
    		messageProperties.setContentType("text/plain");
    		Message message = new Message("mq 消息1234".getBytes(), messageProperties);
    		
    		rabbitTemplate.send("topic001", "mq.abc", message);
    		rabbitTemplate.send("topic002", "rabbit.abc", message);
    	}

**转换器**

1、自定义转换器

*  String 转换器

public class TextMessageConverter implements MessageConverter {
    
    	@Override
    	public Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
    		return new Message(object.toString().getBytes(), messageProperties);
    	}
    
    	@Override
    	public Object fromMessage(Message message) throws MessageConversionException {
    		String contentType = message.getMessageProperties().getContentType();
    		if(null != contentType && contentType.contains("text")) {
    			return new String(message.getBody());
    		}
    		return message.getBody();
    	}
    }

*  PDF

public class PDFMessageConverter implements MessageConverter {
    
    	@Override
    	public Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
    		throw new MessageConversionException(" convert error ! ");
    	}
    
    	@Override
    	public Object fromMessage(Message message) throws MessageConversionException {
    		System.err.println("-----------PDF MessageConverter----------");
    		
    		byte[] body = message.getBody();
    		String fileName = UUID.randomUUID().toString();
    		String path = "d:/010_test/" + fileName + ".pdf";
    		File f = new File(path);
    		try {
    			Files.copy(new ByteArrayInputStream(body), f.toPath());
    		} catch (IOException e) {
    			e.printStackTrace();
    		}
    		return f;
    	}
    
    }

*      Image

public class ImageMessageConverter implements MessageConverter {
    
    	@Override
    	public Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
    		throw new MessageConversionException(" convert error ! ");
    	}
    
    	@Override
    	public Object fromMessage(Message message) throws MessageConversionException {
    		System.err.println("-----------Image MessageConverter----------");
    		
    		Object _extName = message.getMessageProperties().getHeaders().get("extName");
    		String extName = _extName == null ? "png" : _extName.toString();
    		
    		byte[] body = message.getBody();
    		String fileName = UUID.randomUUID().toString();
    		String path = "d:/010_test/" + fileName + "." + extName;
    		File f = new File(path);
    		try {
    			Files.copy(new ByteArrayInputStream(body), f.toPath());
    		} catch (IOException e) {
    			e.printStackTrace();
    		}
    		return f;
    	}
    
    }

2、Json转换器 Jackson2JsonMessageConverter

3、这个会自动转换数据到对应的Java对象

*  指定需要转换的类

MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());
            adapter.setDefaultListenerMethod("consumeMessage");
            Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();
            DefaultJackson2JavaTypeMapper javaTypeMapper = new DefaultJackson2JavaTypeMapper();
            
            Map<String, Class<?>> idClassMapping = new HashMap<String, Class<?>>();
    		idClassMapping.put("order", com.bfxy.spring.entity.Order.class);
    		idClassMapping.put("packaged", com.bfxy.spring.entity.Packaged.class);
    		
    		javaTypeMapper.setIdClassMapping(idClassMapping);
    		
    		jackson2JsonMessageConverter.setJavaTypeMapper(javaTypeMapper);
            adapter.setMessageConverter(jackson2JsonMessageConverter);
            container.setMessageListener(adapter);

*  **对应的Producer**

@Test
    	public void testSendJsonMessage() throws Exception {
    		Order order = new Order();
    		order.setId("001");
    		order.setName("消息订单");
    		order.setContent("描述信息");
    		ObjectMapper mapper = new ObjectMapper();
    		String json = mapper.writeValueAsString(order);
    		System.err.println("order 4 json: " + json);
    		
    		MessageProperties messageProperties = new MessageProperties();
    		//这里注意一定要修改contentType为 application/json
    		messageProperties.setContentType("application/json");
    		Message message = new Message(json.getBytes(), messageProperties);
    		
    		rabbitTemplate.send("topic001", "spring.order", message);
    	}

**4、全局转换器，就是上面的一起配置**

@Bean
        public SimpleMessageListenerContainer messageContainer(ConnectionFactory connectionFactory) {
        	SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        	container.setQueues(queue001(), queue002(), queue003(), queue_image(), queue_pdf());
        	container.setConcurrentConsumers(1);
        	container.setMaxConcurrentConsumers(5);
        	container.setDefaultRequeueRejected(false);
        	container.setAcknowledgeMode(AcknowledgeMode.AUTO);
        	container.setExposeListenerChannel(true);
        	container.setConsumerTagStrategy(new ConsumerTagStrategy() {
    			@Override
    			public String createConsumerTag(String queue) {
    				return queue + "_" + UUID.randomUUID().toString();
    			}
    		});
            
            //全局的转换器:
    		ContentTypeDelegatingMessageConverter convert = new ContentTypeDelegatingMessageConverter();
    		TextMessageConverter textConvert = new TextMessageConverter();
    		convert.addDelegate("text", textConvert);
    		convert.addDelegate("html/text", textConvert);
    		convert.addDelegate("xml/text", textConvert);
    		convert.addDelegate("text/plain", textConvert);
    		
    		Jackson2JsonMessageConverter jsonConvert = new Jackson2JsonMessageConverter();
    		convert.addDelegate("json", jsonConvert);
    		convert.addDelegate("application/json", jsonConvert);
    		
    		ImageMessageConverter imageConverter = new ImageMessageConverter();
    		convert.addDelegate("image/png", imageConverter);
    		convert.addDelegate("image", imageConverter);
    		
    		PDFMessageConverter pdfConverter = new PDFMessageConverter();
    		convert.addDelegate("application/pdf", pdfConverter);
    		
    		adapter.setMessageConverter(convert);
    		container.setMessageListener(adapter);
        	return container;
        }