纯净、安全、绿色的下载网站

首页|软件分类|下载排行|最新软件|IT学院

当前位置:首页IT学院IT技术

SpringBoot使用RabbitMQ 详解SpringBoot中使用RabbitMQ的RPC功能

黑莹de希望   2021-11-20 我要评论
想了解详解SpringBoot中使用RabbitMQ的RPC功能的相关内容吗黑莹de希望在本文为您仔细讲解SpringBoot使用RabbitMQ的相关知识和一些Code实例欢迎阅读和指正我们先划重点:SpringBoot使用RabbitMQ,SpringBoot使用RabbitMQ,RPC下面大家一起来学习吧。

一、RabbitMQ的RPC简介

实际业务中有的时候我们还需要等待消费者返回结果给我们或者是说我们需要消费者上的一个功能、一个方法或是一个接口返回给我们相应的值而往往大型的系统软件生产者跟消费者之间都是相互独立的两个系统部署在两个不同的电脑上不能通过直接对象.方法的形式获取想要的结果这时候我们就需要用到RPC(Remote Procedure Call)远程过程调用方式。
RabbitMQ实现RPC的方式很简单生产者发送一条带有标签(消息ID(correlation_id)+回调队列名称)的消息到发送队列消费者(也称RPC服务端)从发送队列获取消息并处理业务解析标签的信息将业务结果发送到指定的回调队列生产者从回调队列中根据标签的信息获取发送消息的返回结果。

在这里插入图片描述

如图客户端C发送消息指定消息的ID=rpc_id回调响应的队列名称为rpc_resp消息从C发送到rpc_request队列服务端S获取消息业务处理之后将correlation_id附加到响应的结果发送到指定的回调队列rpc_resp中客户端从回调队列获取消息匹配与发送消息的correlation_id相同的值为消息应答结果。

二、SpringBoot中使用RabbitMQ的RPC功能

注意:springboot中使用的时候correlation_id为系统自动生成的reply_to在加载AmqpTemplate实例的时候设置的。

实例:
说明:队列1为发送队列队列2为返回队列

1.先配置rabbitmq

package com.ws.common;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


/*
 * rabbitMQ配置类
 */
@Configuration
public class RabbitMQConfig {
	public static final String TOPIC_QUEUE1 = "topic.queue1";
    public static final String TOPIC_QUEUE2 = "topic.queue2";
    public static final String TOPIC_EXCHANGE = "topic.exchange";
    
    @Value("${spring.rabbitmq.host}")
    private String host;
    @Value("${spring.rabbitmq.port}")
    private int port;
    @Value("${spring.rabbitmq.username}")
    private String username;
    @Value("${spring.rabbitmq.password}")
    private String password;
    
    @Autowired
    ConnectionFactory connectionFactory;
    
    @Bean(name = "connectionFactory")
    public ConnectionFactory connectionFactory() {
    	CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
    	connectionFactory.setHost(host);
    	connectionFactory.setPort(port);
    	connectionFactory.setUsername(username);
    	connectionFactory.setPassword(password);
    	connectionFactory.setVirtualHost("/");
    	return connectionFactory;
    }
    
    @Bean
    public RabbitTemplate rabbitTemplate() {
    	RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
    	//设置reply_to(返回队列只能在这设置)
    	rabbitTemplate.setReplyAddress(TOPIC_QUEUE2);
    	rabbitTemplate.setReplyTimeout(60000);
    	return rabbitTemplate;
    }
    //返回队列监听器(必须有)
    @Bean(name="replyMessageListenerContainer")
    public SimpleMessageListenerContainer createReplyListenerContainer() {
         SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer();
         listenerContainer.setConnectionFactory(connectionFactory);
         listenerContainer.setQueueNames(TOPIC_QUEUE2);
         listenerContainer.setMessageListener(rabbitTemplate());
         return listenerContainer;
    }
    

    
    //创建队列
    @Bean
    public Queue topicQueue1() {
        return new Queue(TOPIC_QUEUE1);
    }
    @Bean
    public Queue topicQueue2() {
        return new Queue(TOPIC_QUEUE2);
    }
    
    //创建交换机
    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange(TOPIC_EXCHANGE);
    }
    
    //交换机与队列进行绑定
    @Bean
    public Binding topicBinding1() {
        return BindingBuilder.bind(topicQueue1()).to(topicExchange()).with(TOPIC_QUEUE1);
    }
    @Bean
    public Binding topicBinding2() {
        return BindingBuilder.bind(topicQueue2()).to(topicExchange()).with(TOPIC_QUEUE2);
    }
}

2.发送消息并同步等待返回值

@Autowired
private RabbitTemplate rabbitTemplate;


//报文body
String sss = "报文的内容";
//封装Message
Message msg = this.con(sss);
log.info("客户端--------------------"+msg.toString());
//使用sendAndReceive方法完成rpc调用
Message message=rabbitTemplate.sendAndReceive(RabbitMQConfig.TOPIC_EXCHANGE, RabbitMQConfig.TOPIC_QUEUE1, msg);
//提取rpc回应内容body
String response = new String(message.getBody());
log.info("回应:" + response);
log.info("rpc完成---------------------------------------------");


public Message con(String s) {
	MessageProperties mp = new MessageProperties();
	byte[] src = s.getBytes(Charset.forName("UTF-8"));
	//mp.setReplyTo("adsdas");   加载AmqpTemplate时设置这里设置没用
	//mp.setCorrelationId("2222");   系统生成这里设置没用
	mp.setContentType("application/json");
	mp.setContentEncoding("UTF-8");
	mp.setContentLength((long)s.length());
	return new Message(src, mp);
} 

3.写消费者

package com.ws.listener.mq;

import java.nio.charset.Charset;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.ws.common.RabbitMQConfig;

import lombok.extern.slf4j.Slf4j;

@Slf4j
@Component
public class Receiver {
	@Autowired
	private RabbitTemplate rabbitTemplate;
	
	@RabbitListener(queues=RabbitMQConfig.TOPIC_QUEUE1)
	public void receiveTopic1(Message msg) {
		log.info("队列1:"+msg.toString());
		String msgBody = new String(msg.getBody());
		//数据处理返回的Message
		Message repMsg = con(msgBody+"返回了", msg.getMessageProperties().getCorrelationId());
		
		rabbitTemplate.send(RabbitMQConfig.TOPIC_EXCHANGE, RabbitMQConfig.TOPIC_QUEUE2, repMsg);
		
    }
	@RabbitListener(queues=RabbitMQConfig.TOPIC_QUEUE2)
	public void receiveTopic2(Message msg) {
		log.info("队列2:"+msg.toString());
		
    }
	
	public Message con(String s, String id) {
		MessageProperties mp = new MessageProperties();
		byte[] src = s.getBytes(Charset.forName("UTF-8"));
		mp.setContentType("application/json");
		mp.setContentEncoding("UTF-8");
		mp.setCorrelationId(id);
		
		return new Message(src, mp);
	} 
}

日志打印:

2019-06-26 17:11:16.607 [http-nio-8080-exec-4] INFO com.ws.controller.UserController - 客户端--------------------(Body:‘报文的内容' MessageProperties [headers={}, contentType=application/json, contentEncoding=UTF-8, contentLength=5, deliveryMode=PERSISTENT, priority=0, deliveryTag=0])

2019-06-26 17:11:16.618 [SimpleAsyncTaskExecutor-1] INFO com.ws.listener.mq.Receiver - 队列1:(Body:‘报文的内容' MessageProperties [headers={}, correlationId=1, replyTo=topic.queue2, contentType=application/json, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=topic.exchange, receivedRoutingKey=topic.queue1, deliveryTag=1, consumerTag=amq.ctag-8IzlhblYmTebqUYd-uferw, consumerQueue=topic.queue1])

2019-06-26 17:11:16.623 [http-nio-8080-exec-4] INFO com.ws.controller.UserController - 回应:报文的内容返回了

2019-06-26 17:11:16.623 [http-nio-8080-exec-4] INFO com.ws.controller.UserController - rpc完成---------------------------------------------


相关文章

猜您喜欢

  • Dubbo分布式服务框架原理机制 java开发分布式服务框架Dubbo原理机制详解

    想了解java开发分布式服务框架Dubbo原理机制详解的相关内容吗Dubbo原理机制详解在本文为您仔细讲解Dubbo分布式服务框架原理机制的相关知识和一些Code实例欢迎阅读和指正我们先划重点:java分布式服务框架,Dubbo原理机制下面大家一起来学习吧。..
  • Python抢购脚本 Python抢购脚本的编写方法

    想了解Python抢购脚本的编写方法的相关内容吗python可乐编程在本文为您仔细讲解Python抢购脚本的相关知识和一些Code实例欢迎阅读和指正我们先划重点:Python抢购脚本,Python抢购脚本写法下面大家一起来学习吧。..

网友评论

Copyright 2020 www.tdogsoftware.com 【零度软件园】 版权所有 软件发布

声明:所有软件和文章来自软件开发商或者作者 如有异议 请与本站联系 点此查看联系方式