「javamq处理」java mq使用

admin 2023-03-21 21:11:08 652

本篇文章给大家谈谈javamq处理，以及java mq使用对应的知识点，希望对各位有所帮助，不要忘了收藏本站喔。

本文目录一览：

1、java怎么将mq接收的文件消息提取出来
2、消息中间件（一）MQ详解及四大MQ比较
3、为什么MQ的编程用JAVA好呢?
4、java 怎么样调用IBM MQ 或者通信问题
5、java事务中调用mq,如果事务回滚,消息会被撤回吗

java怎么将mq接收的文件消息提取出来

WebSphere MQ 接收发送

添加mq jar

类介绍：

SendMSG：消息发送类。

Main（）：主方法。

SendMSG（）：消息发送方法。

方法描述：

package test;

public class SendMSG{

MQEnvironment.hostname = "192.168.10.201";

//通道类型为服务器连接通道

MQEnvironment.channel = "tongdao";

MQEnvironment.CCSID = 1381;

//消息队列端口号

MQEnvironment.port = 10618;

try{

//建立队列管理器QM_SERVER为队列管理器名称

MQQueueManager qMgr = new MQQueueManager("test");

int openOptions = MQC.MQOO_INPUT_AS_Q_DEF|MQC.MQOO_OUTPUTMQC.MQOO_INQUIRE;//建立队列INITQ队列名称INITQ为本地队列

MQQueue queue = qMgr.accessQueue("wanghui",openOptions,null,null,null);

System.out.println("成功建立通道");

MQMessage message = new MQMessage();

message.format = MQC.MQFMT_STRING;

message.characterSet = 1381;

message.writeString("王辉");

message.expiry = -1;//设置消息用不过期

queue.put(message);//将消息放入队列

queue.close();//关闭队列

qMgr.disconnect();//断开连接

}catch(EOFExceptione){

e.printStackTrace();

}catch(MQExceptione){

e.printStackTrace();

}catch(Exceptione){

e.printStackTrace();

}

ReceiveMSG:消息接收类。

Main（）：主方法。

ReceiveMSG():消息接收方法。

public class ReceiveMSG {

MQEnvironment.hostname="192.168.10.201";//通道类型为服务器连接通道

MQEnvironment.channel="tongdao";

MQEnvironment.CCSID=1381;

MQEnvironment.port=10618;

try{

//建立队列管理器QM_SERVER为队列管理器名称

MQQueueManager qMgr = new MQQueueManager("test");

int openOptions=MQC.MQOO_INPUT_AS_Q_DEF|MQC.MQOO_OUTPUT|MQC.MQOO_INQUIRE;//建立队列INITQ队列名称INITQ为本地队列

MQQueue queue=qMgr.accessQueue("wanghui",openOptions,null,null,null);

System.out.println("成功建立通道");

MQMessage message= new MQMessage();

message.format=MQC.MQFMT_STRING;

message.characterSet=1381;

//从队列中获取消息

MQGetMessage Optionspmo=new MQGetMessageOptions();

queue.get(message,pmo);

Stringchars=message.readLine();

System.out.println(chars);

queue.close();//关闭队列

qMgr.disconnect();//断开连接

}catch(EOFExceptione){

e.printStackTrace();

}catch(MQExceptione){

e.printStackTrace();

}catch(Exceptione){

e.printStackTrace();

}

消息中间件（一）MQ详解及四大MQ比较

一、消息中间件相关知识

1、概述

消息队列已经逐渐成为企业IT系统内部通信的核心手段。它具有低耦合、可靠投递、广播、流量控制、最终一致性等一系列功能，成为异步RPC的主要手段之一。当今市面上有很多主流的消息中间件，如老牌的ActiveMQ、RabbitMQ，炙手可热的Kafka，阿里巴巴自主开发RocketMQ等。

2、消息中间件的组成

2.1 Broker

消息服务器，作为server提供消息核心服务

2.2 Producer

消息生产者，业务的发起方，负责生产消息传输给broker，

2.3 Consumer

消息消费者，业务的处理方，负责从broker获取消息并进行业务逻辑处理

2.4 Topic

2.5 Queue

2.6 Message

消息体，根据不同通信协议定义的固定格式进行编码的数据包，来封装业务数据，实现消息的传输

3 消息中间件模式分类

3.1 点对点

PTP点对点:使用queue作为通信载体

说明：

消息生产者生产消息发送到queue中，然后消息消费者从queue中取出并且消费消息。

消息被消费以后，queue中不再存储，所以消息消费者不可能消费到已经被消费的消息。 Queue支持存在多个消费者，但是对一个消息而言，只会有一个消费者可以消费。

说明：

queue实现了负载均衡，将producer生产的消息发送到消息队列中，由多个消费者消费。但一个消息只能被一个消费者接受，当没有消费者可用时，这个消息会被保存直到有一个可用的消费者。

4 消息中间件的优势

4.1 系统解耦

交互系统之间没有直接的调用关系，只是通过消息传输，故系统侵入性不强，耦合度低。

4.2 提高系统响应时间

例如原来的一套逻辑，完成支付可能涉及先修改订单状态、计算会员积分、通知物流配送几个逻辑才能完成；通过MQ架构设计，就可将紧急重要（需要立刻响应）的业务放到该调用方法中，响应要求不高的使用消息队列，放到MQ队列中，供消费者处理。

4.3 为大数据处理架构提供服务

通过消息作为整合，大数据的背景下，消息队列还与实时处理架构整合，为数据处理提供性能支持。

4.4 Java消息服务——JMS

Java消息服务（Java Message Service，JMS）应用程序接口是一个Java平台中关于面向消息中间件（MOM）的API，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。

5 消息中间件应用场景

5.1 异步通信

有些业务不想也不需要立即处理消息。消息队列提供了异步处理机制，允许用户把一个消息放入队列，但并不立即处理它。想向队列中放入多少消息就放多少，然后在需要的时候再去处理它们。

5.2 解耦

降低工程间的强依赖程度，针对异构系统进行适配。在项目启动之初来预测将来项目会碰到什么需求，是极其困难的。通过消息系统在处理过程中间插入了一个隐含的、基于数据的接口层，两边的处理过程都要实现这一接口，当应用发生变化时，可以独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束。

5.3 冗余

有些情况下，处理数据的过程会失败。除非数据被持久化，否则将造成丢失。消息队列把数据进行持久化直到它们已经被完全处理，通过这一方式规避了数据丢失风险。许多消息队列所采用的”插入-获取-删除”范式中，在把一个消息从队列中删除之前，需要你的处理系统明确的指出该消息已经被处理完毕，从而确保你的数据被安全的保存直到你使用完毕。

5.4 扩展性

因为消息队列解耦了你的处理过程，所以增大消息入队和处理的频率是很容易的，只要另外增加处理过程即可。不需要改变代码、不需要调节参数。便于分布式扩容。

5.5 过载保护

在访问量剧增的情况下，应用仍然需要继续发挥作用，但是这样的突发流量无法提取预知；如果以为了能处理这类瞬间峰值访问为标准来投入资源随时待命无疑是巨大的浪费。使用消息队列能够使关键组件顶住突发的访问压力，而不会因为突发的超负荷的请求而完全崩溃。

5.6 可恢复性

系统的一部分组件失效时，不会影响到整个系统。消息队列降低了进程间的耦合度，所以即使一个处理消息的进程挂掉，加入队列中的消息仍然可以在系统恢复后被处理。

5.7 顺序保证

在大多使用场景下，数据处理的顺序都很重要。大部分消息队列本来就是排序的，并且能保证数据会按照特定的顺序来处理。

5.8 缓冲

在任何重要的系统中，都会有需要不同的处理时间的元素。消息队列通过一个缓冲层来帮助任务最高效率的执行，该缓冲有助于控制和优化数据流经过系统的速度。以调节系统响应时间。

5.9 数据流处理

分布式系统产生的海量数据流，如：业务日志、监控数据、用户行为等，针对这些数据流进行实时或批量采集汇总，然后进行大数据分析是当前互联网的必备技术，通过消息队列完成此类数据收集是最好的选择。

6 消息中间件常用协议

6.1 AMQP协议

AMQP即Advanced Message Queuing Protocol,一个提供统一消息服务的应用层标准高级消息队列协议,是应用层协议的一个开放标准,为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同开发语言等条件的限制。

优点：可靠、通用

6.2 MQTT协议

MQTT（Message Queuing Telemetry Transport，消息队列遥测传输）是IBM开发的一个即时通讯协议，有可能成为物联网的重要组成部分。该协议支持所有平台，几乎可以把所有联网物品和外部连接起来，被用来当做传感器和致动器（比如通过Twitter让房屋联网）的通信协议。

优点：格式简洁、占用带宽小、移动端通信、PUSH、嵌入式系统

6.3 STOMP协议

STOMP（Streaming Text Orientated Message Protocol）是流文本定向消息协议，是一种为MOM(Message Oriented Middleware，面向消息的中间件)设计的简单文本协议。STOMP提供一个可互操作的连接格式，允许客户端与任意STOMP消息代理（Broker）进行交互。

优点：命令模式（非topic\queue模式）

6.4 XMPP协议

XMPP（可扩展消息处理现场协议，Extensible Messaging and Presence Protocol）是基于可扩展标记语言（XML）的协议，多用于即时消息（IM）以及在线现场探测。适用于服务器之间的准即时操作。核心是基于XML流传输，这个协议可能最终允许因特网用户向因特网上的其他任何人发送即时消息，即使其操作系统和浏览器不同。

优点：通用公开、兼容性强、可扩展、安全性高，但XML编码格式占用带宽大

6.5 其他基于TCP/IP自定义的协议

有些特殊框架（如：redis、kafka、zeroMq等）根据自身需要未严格遵循MQ规范，而是基于TCP\IP自行封装了一套协议，通过网络socket接口进行传输，实现了MQ的功能。

7 常见消息中间件MQ介绍

7.1 RocketMQ

阿里系下开源的一款分布式、队列模型的消息中间件，原名Metaq，3.0版本名称改为RocketMQ，是阿里参照kafka设计思想使用java实现的一套mq。同时将阿里系内部多款mq产品（Notify、metaq）进行整合，只维护核心功能，去除了所有其他运行时依赖，保证核心功能最简化，在此基础上配合阿里上述其他开源产品实现不同场景下mq的架构，目前主要多用于订单交易系统。

具有以下特点：

官方提供了一些不同于kafka的对比差异：

7.2 RabbitMQ

使用Erlang编写的一个开源的消息队列，本身支持很多的协议：AMQP，XMPP, SMTP,STOMP，也正是如此，使的它变的非常重量级，更适合于企业级的开发。同时实现了Broker架构，核心思想是生产者不会将消息直接发送给队列，消息在发送给客户端时先在中心队列排队。对路由(Routing)，负载均衡(Load balance)、数据持久化都有很好的支持。多用于进行企业级的ESB整合。

7.3 ActiveMQ

Apache下的一个子项目。使用Java完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现，少量代码就可以高效地实现高级应用场景。可插拔的传输协议支持，比如：in-VM, TCP, SSL, NIO, UDP, multicast, JGroups and JXTA transports。RabbitMQ、ZeroMQ、ActiveMQ均支持常用的多种语言客户端 C++、Java、.Net,、Python、 Php、 Ruby等。

7.4 Redis

使用C语言开发的一个Key-Value的NoSQL数据库，开发维护很活跃，虽然它是一个Key-Value数据库存储系统，但它本身支持MQ功能，所以完全可以当做一个轻量级的队列服务来使用。对于RabbitMQ和Redis的入队和出队操作，各执行100万次，每10万次记录一次执行时间。测试数据分为128Bytes、512Bytes、1K和10K四个不同大小的数据。实验表明：入队时，当数据比较小时Redis的性能要高于RabbitMQ，而如果数据大小超过了10K，Redis则慢的无法忍受；出队时，无论数据大小，Redis都表现出非常好的性能，而RabbitMQ的出队性能则远低于Redis。

7.5 Kafka

Apache下的一个子项目，使用scala实现的一个高性能分布式Publish/Subscribe消息队列系统，具有以下特性：

7.6 ZeroMQ

号称最快的消息队列系统，专门为高吞吐量/低延迟的场景开发，在金融界的应用中经常使用，偏重于实时数据通信场景。ZMQ能够实现RabbitMQ不擅长的高级/复杂的队列，但是开发人员需要自己组合多种技术框架，开发成本高。因此ZeroMQ具有一个独特的非中间件的模式，更像一个socket library，你不需要安装和运行一个消息服务器或中间件，因为你的应用程序本身就是使用ZeroMQ API完成逻辑服务的角色。但是ZeroMQ仅提供非持久性的队列，如果down机，数据将会丢失。如：Twitter的Storm中使用ZeroMQ作为数据流的传输。

ZeroMQ套接字是与传输层无关的：ZeroMQ套接字对所有传输层协议定义了统一的API接口。默认支持进程内(inproc) ，进程间(IPC) ，多播，TCP协议，在不同的协议之间切换只要简单的改变连接字符串的前缀。可以在任何时候以最小的代价从进程间的本地通信切换到分布式下的TCP通信。ZeroMQ在背后处理连接建立，断开和重连逻辑。

特性：

二、主要消息中间件的比较

为什么MQ的编程用JAVA好呢?

作为 MQSeries 应用程序的开发人员，MQSeries Java 编程接口使您能够享用到许多 Java 的优点。

Java 编程语言易于使用。它不需要头文件、指针、结构、联合以及运算符重载。用 Java 写的程序比其同类 C 和 C++ 更容易开发调试。

Java 是面向对象的。Java 的面向对象功能好比 C++，但是它没有多重继承性。实际上，Java 使用的是接口的概念。

Java 是继承性分布的。Java 类库包含了使用 TCP/IP 的协议（如 HTTP, FTP）的例行程序库。 Java 程序能够象访问文件系统那样容易地访问 URL。

Java 是健壮的。Java 非常着重于可能发生问题的早期检查，动态（运行时）检查并消除有出错可能性的情况。 Java 使用引用的概念，它能够消除覆盖内存和破坏数据的可能性。

Java 是安全的。Java 要在网络/分布式环境中运行，并已在安全性问题上下了很大功夫。 Java 程序不能超越其运行时堆栈，不能破坏超过其处理空间的内存，并且当从 Internet 下载时，还不能读写本地文件。

Java 程序是可移植的。Java 规范的特征中没有 “实现相关性”这一条。Java 编译器生成一个体系结构中性的目标文件格式。编译代码可在许多处理器上执行，只要有 Java 运行时系统。

若用 MQSeries Java 编写应用程序，用户可从 Internet 为程序（称为小程序）下载 Java 字节代码，并在其自用的机器上运行。这意味着访问 Web 服务器的用户不需要先在其机器上安装，就能装入并运行应用程序。需要更新程序时，可在 Web 服务器上更新复制，当用户下一次访问小程序时，它们会收到最新的版本。这能够显著地降低成本，而安装并更新传统的客户程序需要涉及大量的桌面。若将小程序置于在企业防火墙外可存取的 Web 服务器上，则无论谁在 Internet 上都可以下载并使用您的应用程序。这代表 MQSeries 系统可从 Internet 上的任何地方获得信息。这打开了建立一组全新的完整 Internet 可存取服务的大门，支持商用程序并使之电子化。

java 怎么样调用IBM MQ 或者通信问题

websphere mq : 用于传输信息具有跨平台的功能。

1 安装websphere mq 并启动

2 websphere mq 建立 queue Manager (如：MQSI_SAMPLE_QM)

3 建立queue 类型选择 Local类型的（如lq ）

3 建立channels 类型选择Server Connection （如BridgeChannel）

java 代码如下：

package test.mq;

import com.ibm.mq.*;

* 成功的访问mq 的java 类

public class FirstMqTest {

// public static void main(String[] args[]){

// FirstMqTest first = new FirstMqTest();

// first.test();

// }

public static void main(String args[]){

FirstMqTest first = new FirstMqTest();

first.test();

}

public void test(){

String qManager = "MQSI_SAMPLE_QM"; //QueueManager name

String qName = "lq";//Queue Name

try {

//configure connection parameters

MQEnvironment.hostname="172.16.17.123";//MQ Server name or IP

//MQEnvironment.port=1414;//listenr port

MQEnvironment.channel="BridgeChannel";//Server-Connection Channel

MQEnvironment.CCSID =1381;

// Create a connection to the QueueManager

System.out.println("Connecting to queue manager: "+qManager);

MQQueueManager qMgr = new MQQueueManager(qManager);

// Set up the options on the queue we wish to open

int openOptions = MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_OUTPUT;

// Now specify the queue that we wish to open and the open options

System.out.println("Accessing queue: "+qName);

MQQueue queue = qMgr.accessQueue(qName, openOptions);

// Define a simple WebSphere MQ Message ...

MQMessage msg = new MQMessage();

// ... and write some text in UTF8 format

msg.writeUTF("Hello, World!");

// Specify the default put message options

MQPutMessageOptions pmo = new MQPutMessageOptions();

// Put the message to the queue

System.out.println("Sending a message...");

* 在此测试一下 mq 的传输次列

for(int j=0;j 5;j++){

String str ="test11111111111";

str = str+j;

msg.writeUTF(str);

queue.put(msg, pmo);

}

queue.put(msg, pmo);

// Now get the message back again. First define a WebSphere MQ message

// to receive the data

MQMessage rcvMessage = new MQMessage();

// Specify default get message options

MQGetMessageOptions gmo = new MQGetMessageOptions();

// Get the message off the queue.

System.out.println("...and getting the message back again");

queue.get(rcvMessage, gmo);

// And display the message text...

String msgText = rcvMessage.readUTF();

System.out.println("The message is: " + msgText);

// Close the queue

System.out.println("Closing the queue");

queue.close();

// Disconnect from the QueueManager

System.out.println("Disconnecting from the Queue Manager");

qMgr.disconnect();

System.out.println("Done!");

}

catch (MQException ex) {

System.out.println("A WebSphere MQ Error occured : Completion Code "

+ ex.completionCode + " Reason Code " + ex.reasonCode);

}

catch (java.io.IOException ex) {

System.out.println("An IOException occured whilst writing to the message buffer: "

+ ex);

}

java事务中调用mq,如果事务回滚,消息会被撤回吗

回。java事务中调用mq是公司开发的平台程序，程序内设置了撤回语言，如果事务回滚消息会被撤回处理，Java的事务处理，如果对数据库进行多次操作，每一次的执行或步骤都是一个事务。

关于javamq处理和java mq使用的介绍到此就结束了，不知道你从中找到你需要的信息了吗？如果你还想了解更多这方面的信息，记得收藏关注本站。

标签：javamq处理