Rabbitmq中得RPC调用代码详解

server/2024/10/22 15:33:24/

文章目录

  • 1.RPC客户端
  • 2.RabbitMQ连接信息实体类
  • 3.XML工具类

本文档只是为了留档方便以后工作运维，或者给同事分享文档内容比较简陋命令也不是特别全，不适合小白观看，如有不懂可以私信，上班期间都是在得

直接上代码了

1.RPC客户端

RPC客户端

/*** @ClassName: RPCClient* @Description: RPC 客户端* @Author: XHao* @Date: 2024/8/30 11:14*/
public class RPCClient {private Connection connection;private Channel channel;private String requestQueueName = "XYG.HS.MES.PRD.CNXsvr";private String replyQueueName;public RPCClient() {}public RPCClient(XygMqIesConnInfo xygMqIesConnInfo) throws IOException, TimeoutException {//建立一个连接和一个通道，并为回调声明一个唯一的'回调'队列ConnectionFactory factory = new ConnectionFactory();factory.setHost(xygMqIesConnInfo.getHost());factory.setPort(xygMqIesConnInfo.getPort());factory.setUsername(xygMqIesConnInfo.getUserName());factory.setPassword(xygMqIesConnInfo.getPwd());factory.setVirtualHost("IES");try {connection = factory.newConnection();System.err.println("===============创建通道===============");channel = connection.createChannel();System.err.println("===============创建成功===============");}catch (Exception e){System.err.println("报错信息=============="+e.getMessage());}//定义一个临时变量的接受队列名System.err.println("===============定义一个临时变量的接受队列名===============");replyQueueName = channel.queueDeclare().getQueue();}//发送RPC请求public String call(String message) throws IOException, InterruptedException {//生成一个唯一的字符串作为回调队列的编号String corrId = UUID.randomUUID().toString();//发送请求消息，消息使用了两个属性：replyto和correlationId//服务端根据replyto返回结果，客户端根据correlationId判断响应是不是给自己的AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().correlationId(corrId).replyTo(replyQueueName).build();//发布一个消息，requestQueueName路由规则System.err.println("===============发布一个消息===============");System.err.println("===============消息内容===============");System.err.println("==============="+ message + "===============");System.err.println("===============================");System.err.println("===============================");System.err.println("===============================");System.err.println("===============回调队列的编号===============");System.err.println("===============请求时间："+new Date());System.err.println("==============="+ "correlationId::"+props.getCorrelationId() + "===============");System.err.println("==============="+ "ReplyTo::"+props.getReplyTo() + "===============");channel.basicPublish("", requestQueueName, props, message.getBytes(StandardCharsets.UTF_8));//由于我们的消费者交易处理是在单独的线程中进行的，因此我们需要在响应到达之前暂停主线程。//这里我们创建的 容量为1的阻塞队列ArrayBlockingQueue，因为我们只需要等待一个响应。final BlockingQueue<String> response = new ArrayBlockingQueue<String>(1);//获取响应消息System.err.println("===============获取响应消息===============");channel.basicConsume(replyQueueName, true, new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,byte[] body) throws IOException {//检查它的correlationId是否是我们所要找的那个if (properties.getCorrelationId().equals(corrId)) {//如果是，则响应BlockingQueueresponse.offer(new String(body, "UTF-8"));}}});return response.take();}public void close() throws IOException {connection.close();}public static JSONObject getResult(XygMqIesConnInfo xygMqIesConnInfo, String msg) {if (Objects.isNull(xygMqIesConnInfo) || Objects.isNull(msg)) {return null;}RPCClient fibonacciRpc = null;String response = null;try {fibonacciRpc = new RPCClient(xygMqIesConnInfo);response = fibonacciRpc.call(msg);System.err.println("响应消息+"+response);} catch (IOException | InterruptedException e) {e.printStackTrace();throw new RuntimeException("RPC调用异常");} catch (TimeoutException e) {e.printStackTrace();throw new RuntimeException("RPC调用超时");} finally {if (fibonacciRpc != null) {try {fibonacciRpc.close();} catch (IOException ignore) {}}}return XmlUtil.xmlToJson(response);}
}

2.RabbitMQ连接信息实体类

RabbitMQ连接信息实体类

/*** @ClassName: XygMqIESConnectionInfo* @Description: RabbitMQ连接信息* @Author: XHao* @Date: 2024/8/22 17:07*/
@Data
@JsonInclude(JsonInclude.Include.NON_NULL)
@TableName("xyg_mq_ies_conn_info")
@ApiModel(value = "连接信息对象", description = "IES RabbitMQ连接信息表")
public class XygMqIesConnInfo {private static final long serialVersionUID = 1L;@TableId(value = "ID", type = IdType.AUTO)private Long id;@ApiModelProperty(value = "园区ID")private String parkId;@ApiModelProperty(value = "车间编码")private String workshopCode;@ApiModelProperty(value = "主机地址")private String host;@ApiModelProperty(value = "端口")private Integer port;@ApiModelProperty(value = "用户名")private String userName;@ApiModelProperty(value = "密码")private String pwd;@ApiModelProperty(value = "队列名称")private String queueName;@ApiModelProperty(value = "交换机名称")private String exchangeName;@ApiModelProperty(value = "路由键")private String routingKey;
}

3.XML工具类

XML工具类

/*** @ClassName: XmlUtil* @Description: xml 解析与生成工具类* @Author: XHao* @Date: 2024/8/20 14:28*/
public class XmlUtil {/*** XML节点转换JSON对象** @param element 节点* @param object  新的JSON存储* @return JSON对象*/private static JSONObject xmlToJson(Element element, JSONObject object) {List<Element> elements = element.elements();for (Element child : elements) {Object value = object.get(child.getName());Object newValue;if (child.elements().size() > 0) {JSONObject jsonObject = xmlToJson(child, new JSONObject(true));if (!jsonObject.isEmpty()) {newValue = jsonObject;} else {newValue = child.getText();}} else {newValue = child.getText();}List<Attribute> attributes = child.attributes();if (!attributes.isEmpty()) {JSONObject attrJsonObject = new JSONObject();for (Attribute attribute : attributes) {attrJsonObject.put(attribute.getName(), attribute.getText());attrJsonObject.put("content", newValue);}newValue = attrJsonObject;}if (newValue != null) {if (value != null) {if (value instanceof JSONArray) {((JSONArray) value).add(newValue);} else {JSONArray array = new JSONArray();array.add(value);array.add(newValue);object.put(child.getName(), array);}} else {object.put(child.getName(), newValue);}}}return object;}/*** XML字符串转换JSON对象** @param xmlStr XML字符串* @return JSON对象*/public static JSONObject xmlToJson(String xmlStr) {JSONObject result = new JSONObject(true);SAXReader xmlReader = new SAXReader();try {Document document = xmlReader.read(new StringReader(xmlStr));Element element = document.getRootElement();return xmlToJson(element, result);} catch (Exception e) {e.printStackTrace();}return result;}/*** XML文件转换JSON对象** @param xmlString xml字符串* @param node      选择节点* @return JSON对象*/public static JSONObject xmlToJson(String xmlString, String node) {JSONObject result = new JSONObject(true);SAXReader xmlReader = new SAXReader();try {//将给定的String文本解析为XML文档并返回新创建的documentorg.dom4j.Document document = DocumentHelper.parseText(xmlString);
//            Document document = xmlReader.read(file);Element element;if (StringUtils.isBlank(node)) {element = document.getRootElement();} else {element = (Element) document.selectSingleNode(node);}return xmlToJson(element, result);} catch (Exception e) {e.printStackTrace();}return result;}/*** 生成xml格式的字符串** @return*/public static String createXmlString(XmlParam xmlParam) {//创建document对象org.dom4j.Document document = DocumentHelper.createDocument();//设置编码document.setXMLEncoding("UTF-8");//创建根节点Element message = document.addElement("Message");// 开始组装 Header 节点// 在 Header 节点下加入子节点Element header = message.addElement("Header");// 组装固定值for (HeaderEnum h : HeaderEnum.values()) {Element childNode = header.addElement(h.name());childNode.setText(h.getValue());}// 组装传参值Map<String, String> headerMap = JSONObject.parseObject(JSONObject.toJSONString(xmlParam.getHeader()), Map.class);headerMap.forEach((k, v) -> {Element childNode = header.addElement(k.toUpperCase());childNode.setText(v);});// 组装事务ID，唯一值:当前时间戳Element transactionId = header.addElement("TRANSACTIONID");SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");transactionId.setText(sdf.format(new Date()) + String.valueOf(Calendar.getInstance().getTimeInMillis()));Element listener = header.addElement("listener");listener.setText("QueueListener");// 开始组装 Body 节点Element body = message.addElement("Body");Map<String, String> bodyMap = JSONObject.parseObject(JSONObject.toJSONString(xmlParam.getBody()), Map.class);bodyMap.forEach((k, v) -> {if (Objects.isNull(v)) {return;}Element childNode = body.addElement(k.toUpperCase());childNode.setText(v);});//将document对象转换成字符串String xml = document.asXML();// 去掉 XML 声明if (xml.startsWith("<?xml")) {xml = xml.substring(xml.indexOf(">") + 1);}return xml;}

如果点赞多，评论多会更新详细教程，待补充。


http://www.ppmy.cn/server/116125.html

相关文章

Java 日志

Java 日志

日志就是为了将程序的运行状况保存到文件中去。 命名的一个小细节&#xff1a; 比如把信息保存到文件中这个方法的名字可以写为infoToFile&#xff0c;有个人为了偷懒&#xff0c;写成info2File&#xff0c;发现效果还挺好&#xff0c;一下就能分清两个单词&#xff0c;所以后…
阅读更多...
【网络安全】-rce漏洞-pikachu

【网络安全】-rce漏洞-pikachu

rce漏洞包含命令执行漏洞与代码执行漏洞 文章目录 前言 什么是rce漏洞&#xff1f; 1.rce漏洞产生原因&#xff1a; 2.rce的分类&#xff1a; 命令执行漏洞&#xff1a; 命令拼接符&#xff1a; 常用函数&#xff1a; 代码执行漏洞&#xff1a; 常用函数&#xff1a; 分类&…
阅读更多...
【开源免费】基于SpringBoot+Vue.JS在线视频教育平台（JAVA毕业设计）

【开源免费】基于SpringBoot+Vue.JS在线视频教育平台（JAVA毕业设计）

本文项目编号 T 027 &#xff0c;文末自助获取源码 \color{red}{T027&#xff0c;文末自助获取源码} T027&#xff0c;文末自助获取源码 目录 一、系统介绍二、演示录屏三、启动教程四、功能截图五、文案资料5.1 选题背景5.2 国内外研究现状5.3 可行性分析 六、核心代码6.1 新…
阅读更多...
微软数据库的SQL注入漏洞解析——Microsoft Access、SQLServer与SQL注入防御

微软数据库的SQL注入漏洞解析——Microsoft Access、SQLServer与SQL注入防御

说明：本文仅是用于学习分析自己搭建的SQL漏洞内容和原理，请勿用在非法途径上，违者后果自负，与笔者无关；本文开始前请认真详细学习《‌中华人民共和国网络安全法》‌及其相关法规内容【学法时习之丨网络安全在身边一图了解网络安全法_中央网络安全和信息化委员会办公室】 。…
阅读更多...
记忆化搜索【下】

记忆化搜索【下】

375. 猜数字大小II 题目分析 题目链接&#xff1a;375. 猜数字大小 II - 力扣&#xff08;LeetCode&#xff09; 题目比较长&#xff0c;大致意思就是给一个数&#xff0c;比如说10&#xff0c;定的数字是7&#xff0c;让我们在[1, 10]这个区间猜。 如果猜大或猜小都会说明…
阅读更多...
拱式桥安全结构健康监测解决方案

拱式桥安全结构健康监测解决方案

拱式桥作为一种常见的桥梁结构&#xff0c;其拱形设计不仅美观&#xff0c;还具有较高的承载能力。然而&#xff0c;随着使用年限的增加和环境因素的影响&#xff0c;拱式桥的结构健康和稳定需要持续监测和评估。自动化监测技术的应用&#xff0c;可以提升拱式桥的监测效率和准…
阅读更多...
程易科技AI OS：赋能开发者，构建智慧未来

程易科技AI OS：赋能开发者，构建智慧未来

【引言】 随着人工智能技术的迅猛发展&#xff0c;越来越多的企业和个人投身于AI应用的研发之中。在这个过程中&#xff0c;一套高效、灵活且功能强大的开发平台显得尤为重要。程易科技推出的人工智能操作系统&#xff08;AI OS&#xff09;&#xff0c;正是为了满足这一市场需…
阅读更多...
Docker部署tenine实现后端应用的高可用与负载均衡

Docker部署tenine实现后端应用的高可用与负载均衡

采用Docker方式的Tengine 和 keepalived 组合模式可以实现小应用场景的高可用负载均衡需求 目录 网络架构一、环境准备二、软件安装1. 下载Tenine镜像2. 下载Keepalived镜像3. 制作SpringBoot镜像 三、软件配置1. 创建应用容器2. 代理访问应用3. 创建Keepalived4. 测试高可用 网…
阅读更多...
最新文章

Copyright @ 2022~2023