对于阿里中间件技术：应用服务器篇感兴趣的读者，本文将提供您所需要的所有信息，我们将详细讲解阿里中间件，并且为您提供关于10Gmysqlbinlog重放并传输到另一台服务器执行，阿里中间件大赛、2018

对于阿里中间件技术：应用服务器篇感兴趣的读者，本文将提供您所需要的所有信息，我们将详细讲解阿里 中间件，并且为您提供关于10G mysql binlog重放并传输到另一台服务器执行，阿里中间件大赛、2018年阿里中间件性能挑战赛--个人总结、C++ Web应用服务器中间件 MYCP 全面介绍、ESB 系列之中间件技术入门教程的宝贵知识。

本文目录一览：

• 阿里中间件技术：应用服务器篇（阿里 中间件）
• 10G mysql binlog重放并传输到另一台服务器执行，阿里中间件大赛
• 2018年阿里中间件性能挑战赛--个人总结
• C++ Web应用服务器中间件 MYCP 全面介绍
• ESB 系列之中间件技术入门教程

阿里中间件技术：应用服务器篇（阿里 中间件）

10G mysql binlog重放并传输到另一台服务器执行，阿里中间件大赛

转载自：https://tianchi.aliyun.com/forum/new_articleDetail.html?spm=5176.11165310.0.0.90a57f61Sy5xTQ&raceId=231600&postsId=2035

这个冠军的方案确实赞，10G的mysql binlog重放并传输只用了2秒！

总决赛冠军队伍 作死小分队 比赛攻略

 

决赛答辩PPT已上传！点这里查看。

赛题分析

给定一批固定的增量数据变更信息（存放在Server端），程序需要单线程顺序读取文件，进行数据重放计算，然后将最终结果输出到给定的目标文件（在Client端）中。

增量数据的变更主要包含数据库的Insert/Update/Delete三种类型的数据。主键可能发生变更。

为了降低数据在网络中的传输开销，我们的设计是在Server端完成数据的重放计算，再将结果发送到Client端，写入结果文件中。

比赛采用16核机器运行程序。显然，我们需要设计一个并行算法，充分利用多核CPU。

解题思路

由于数据的重放计算在Server端，Client主要负责接收结果和写入文件，因此核心的算法都在Server端。本节将首先给出Server端算法的系统架构，之后具体介绍多线程算法的细节与实现。

系统架构

为了最大限度利用多核CPU的计算能力，我们将整个重放过程按照流水线的方式分成三个部分：

Fig.1 系统架构图

1) Reader 按照赛题要求，负责单线程读取文件。事实上，由于单线程的内存拷贝速度已经跟不上流水线速度，我们用 MappedByteBuffer 将文件按照16M的大小切分成多个的分段（Segment）映射到内存，交由Parser处理。

为防止最后的一个 Task 被从中间切成两段，Segment 除了 16MB 数据以外还多保留了 128KB 的 Margin。

2) Parser 负责从 Segment 中初步 Parse 出 Task的操作类型 (Insert/Update/Delete) 以及主键值。根据主键值，将 Task 分成 N 个 Bucket 分别交给 Worker 处理。 Parser支持多线程执行，可伸缩。

如果是 UpdatePK, 则产生 UpdatePKSrc 和 UpdatePKDst 两个 task 以及对应的 Promise 对象，按照各自的 PK , 分别发给对应的 Bucket (Promise对象及Bucket之间如何协作处理主键更新问题下文会详细说明)。

Parser 产生的 task 不是一个一个发给 Worker/Bucket 的。这样吞吐量上不去。更好的方案是按批发送。如图所示，每个 Segment 对于每个 Bucket 产生一批 tasks，这些 tasks 被附加到对应的 Segment 上，供 Worker 读取。

容易误解的一点：Parser 并不是 Parse 全部的 key-value，这不够高效。Parser 只负责第一列也就是主键，剩下的部分，通过把当前的 offset 传给 Worker，从而交给 Worker 来处理。

3) Worker 和 Bucket 是一一对应关系，Worker 根据 Parser 产生的结果，在自己的 Bucket 上依次重放 task，多线程执行，可伸缩。

当所有 Worker 处理完最后的 task 时，意味着回放完成，可以准备输出了。输出其实是 merge K 个有序 stream 的经典问题，可以用堆来高效的解决。

上述的流水线非常适合用 RingBuffer 实现，原因如下：

* RingBuffer 相比 BlockingQueue 速度更快，且本身不产生新对象，减少 GC

* RingBuffer 能够方便地为 slot 静态分配内存空间

这里我们用了 Disruptor 框架，它是一个高性能的线程间消息通讯的库，底层用 RingBuffer 实现。它不仅能取代 ArrayBlockingQueue，功能上还要丰富的多。对于本题的架构，只需要一个 RingBuffer 就能完成。

以下的示意图是和 Fig.1 是 完全等价 的：

Fig.2 RingBuffer流水线效果图

图中白色部分为当前RingBuffer内的数据。图中 Worker 2 和 Worker 0 都在处理 Segment 103 中对应自己 Bucket 的 task。

并行算法

为了发挥并行性能，在每个Parser中我们将数据表按 hash(PK) 切分成 N 个 Bucket, 每个Bucket都由一个独立的Worker线程完成重放计算。

Fig.3 Hash 分桶示意图

对于Insert，Delete 和一般的 Update 事件，只要分配到对应的 Bucket 去做就可以了；唯独 UpdatePK（更新主键）事件例外，必须要 Bucket 间协作才能保证正确地把数据移动过去。

何谓“Bucket 间协作”？

一行数据被“拿走”的时候，如果还存在对该行的操作没完成，那这些修改就丢失了！所以，必须要保证一行数据被“拿走”前，所有的修改都已经 apply 到上面。反之同理，必须要“拿到”数据以后，才能把后续的操作 apply 上去。

也就是说，我们要让这两个线程在这一点（UpdatePK 这条操作）上同步才行。

Fig.4 UpdatePK 对 Bucket 的影响，接收方要等待发送方也处理到这个task

很自然的想到，可以用 CountDownLatch 来阻塞 UpdatePK 的接收方（数据被移到此 Bucket），直到 UpdatePK 的发送方发出这行数据，它才拿到数据、接着运行。然而，当 UpdatePK 操作较为密集的时候，这个解决方案非常低效！

另一种思路是内存中维护一张主键变更表，记录主键的变更历史，将所有 UpdatePK 后新主键的所有操作都分配到旧主键所在的Bucket中。然而每个task分配时都需要从主键变更表中查找对应的Bucket，并且Parser也无法并行执行，同样十分低效。

有什么更高效的解决方法呢?

针对这一问题, 最终设计出下文的并行算法, 本算法的核心在于通过 Promise 对象，解决了 Update 主键这一难题，从而使得数据表的各个 Bucket 的线程能够无锁、高效地协作。

所谓 Promise，是借鉴 Future/Promise 异步编程范式设计的一个数据结构。实现很简单，只要封装一个 volatile 的变量，如下所示（实际代码实现更复杂，仅为示例）：

final class Promise {
    volatile T data;

    public boolean ready() { return data != null; }
    public T get() { return data; }
    public void set(T data) { this.data = data; }
}

Promise 在 Parser parse 到 UpdatePK 事件时产生，发送方和接收方都持有它的引用：发送方获得 UpdatePKSrc 任务，只能写入Promise（set）；接收方获得 UpdatePKDst 任务，只能读取Promise（get）、以及检查数据是否 ready。通过 Promise 作为中间媒介，被操作的数据记录 data 就能从源 Bucket：hash(srcKey) 移动到目标 Bucket：hash(dstKey)。

Fig.5 Promise与发送方和接收方的关系

相比上一小节提到的 Latch 解决方案，Promise 不是阻塞接收方，而是告诉他：你要的数据还没准备好。明智的接收方将会先“搁置”这个消息，并把后来遇到的所有对这个 Key 的操作都暂存起来（放在blockedTable中，如图所示）。一旦某一时刻 Promise.ready() 为真，就可以把这个 data 放到对应的 Key 上了！暂存的操作也可以那时候再做。

如何从 blockedTable 中高效地找到 promise.ready() 的 task？事实上对于每个发送者（也就是其他bucket），我们只要检查最早阻塞的那个 task 是否 ready 就可以了。

Fig.6 Task 处理效果图，暂存被阻塞的Task

对于被 block 的 PK 的操作，会以一个链表保存下来。如果不巧操作很多，这个链表就会变的很长。一个简单的改进：如果来的是一个普通 Update 操作，其实可以直接 apply 到上一个操作上的 data 上，例如 A=5 B=6 可以叠加到 PK=3 A=4上， 就变成 PK=3 A=5 B=6，从而避免把 Update 操作追加到链表上。

附上关键逻辑的伪代码：

// 当 Parse 到 UpdatePK (SET PK=dstKey WHERE PK=srcKey)
Promise promise = new Promise();
bucket[hash(srcKey)].send(new UpdatePKSrcTask(srcKey, promise, ...));
bucket[hash(dstKey)].send(new UpdatePKDstTask(dstKey, promise, ...));

// 当接收方 Bucket 收到 UpdatePKDstTask task
if (!task.promise.ready()) {
    LinkedList<Task> blockedTasksOnThisKey = new LinkedList<>(); // 存放阻塞的Task
    blockedTasksOnThisKey.add(task);
    blockedTable.put(task.key, blockedTasksOnThisKey); // 暂存，以后再处理
} else {
    table.put(task.key, task.getData()); // 直接处理
}

// 当发送方 Bucket 收到 UpdatePKSrcTask task
task.promise.set(table.remove(key));

// 当接受方发现 blockedTable 中的 task.promise 已经 ready，则取出来处理掉
for (Task task : blockedTasks) {
    applyUnblockedTask(task);
}

如果阻塞的 Tasks 中包含一个 Delete，后面又来了一个 UpdatePKDst，要注意，可能会再次阻塞。

您可能担心 blockedTable 的查询增加了单个 Bucket 的计算负担。实验表明，由于各个 Bucket 的工作进度差异相差不会很大，blockedTable 的最大 size 也在 25000 以内，远小于数据表大小，所以这个代价是完全可接受的。

健壮性分析

算法的正确性是最重要的。对于任意的输入数据集，算法都能保证输出正确的结果。

可以从理论上证明：本算法可以处理以任何顺序出现的 UpdatePK / Update / Delete / Insert 操作，保证重放结束后一定查询到正确的结果。

其实很简单，算法保证了所有的操作都在它们可以执行的时候被执行，换句话说，对于一切有互相依赖关系的操作，算法不会破坏它们的先后关系。算法的并行性，是在保证了该前提的情况下做到的。

以下的例子可能帮助您获得一个感性的认识。

从一个简单的例子开始。如果遇到如下的序列

Insert PK=1 A=1 B=2
Update A=2 Where PK=1
Update PK=2 Where PK=1
Update A=3 Where PK=2

假设 PK = 1 和 2 分别被 hash 到 Bucket 1 和 Bucket 2，那么会有如下情况：

Insert PK=1 A=1 B=2    // Bucket 1 新增记录
Update A=2 Where PK=1  // Bucket 1 更新记录 
Update PK=2 Where PK=1 // Bucket 1 接到 UpdatePKSrc，移除并把该数据 set 到 promise
                       // Bucket 2 如果拿到了 data，那就成功了；假设它没拿到，PK=2加入阻塞表
Update A=3 Where PK=2  // Bucket 2 把这条 Update 追加到 PK=2 的操作列表上
                          ...
                       // Bucket 2 等到了 data，Update A=3 也被重放了

举一个更极端的例子来说明。如果遇到这样的情况：

Insert PK=1 A=1 B=2
Update A=2 Where PK=1
Update PK=2 Where PK=1
Update PK=3 Where PK=2 // 连续的更新
Update A=3 Where PK=3

算法会做如下处理：

Insert PK=1 A=1 B=2    // Bucket 1 新增记录
Update A=2 Where PK=1  // Bucket 1 更新记录 
Update PK=2 Where PK=1 // Bucket 1 接到 UpdatePKSrc，移除并把该数据 set 到 promise
                       // Bucket 2 如果拿到了 data，那就成功了；假设它没拿到，PK=2加入阻塞表
Update PK=3 Where PK=2 // Bucket 2 发现 2 这个主键在阻塞表中，所以本操作也放入阻塞表
                       // Bucket 3 无法拿到 data，所以把 PK=3 加入阻塞表
Update A=3 Where PK=3  // Bucket 3 把这条 Update 追加到 PK=3 的操作列表上
                          ...
                       // Bucket 2 等到了 data，UpdatePK 也被重放了
                          ...
                       // Bucket 3 等到了 data，Update A=3 也被重放了

不妨自己尝试更多的情况。

 

关于表结构的健壮性, 程序会根据第一次遇到的 Insert log 来确定表结构，包括各个列的名字、类型、主键信息等。

 

程序严格按照比赛要求。对于数字，支持 long 型正数范围；对于文本，最长支持 65536 个字符。具体实现参考下文“数据存储”一小节。

优势与创新点

在健壮性的基础上，本算法还有以下几点优势：

* 完全无锁（Lock-free），无阻塞（Non-blocking）。在16核CPU的测试场景下，锁竞争将会导致不小的开销；而阻塞更不用说，极端情况下可能多线程会退化成协程。（例如 Latch 的解决方案，连续 UpdatePK 就会导致这样的情况）。本算法完全摈弃了wait()或lock()，而是用 代价极低 的 volatile 实现同步，这是最大的创新点。

* 可伸缩（Scalability）。除了 Reader 根据题意必须单线程，算法中没有任何不可伸缩的数据结构，理论上为线性加速比。若 CPU 核数增加，只要提升 Parser 和 Worker/Bucket 的线程数即可。（一些解决方案用到全局的 KeyMap，导致无法伸缩）

* 流处理（Streaming）。本算法是一个真正的流处理系统，在真实场景中可以不断灌入新数据并提供查询（保证最终一致性）。这也与比赛的初衷一致。

细节实现和优化

上述算法和架构给出了大致的代码编写思路。细节上，为了追求极致的性能，我们还做了各种优化。

原生类型数据结构

Java 的范型对于 primitive type 的数据是严重的浪费。比如 Map 是非常低效的，不仅浪费了大量内存，还产生了大量冗余的 boxing/unboxing。

对此，我们利用 fastutil 和 koloboke 这两个库代替 Java 标准库中范型实现的 HashMap、ArrayList 等数据结构，极大提升了性能。

数据存储

我们使用 long 数组来存储每一行数据。

由于列值的类型为 long 或者 String。对于 long 类型值, 将其解析成 long 数据存储即可。而对于String 类型的数据, 如果通过将其转换成 String 存储, 至少有两个问题：

• Encode/decode 造成无谓的性能损耗；
• 内存开销很大，对象数量非常多，对 GC 造成巨大压力。

StringStore 类能够将 String 类型的列也“变成” long，从而放到 long 数组中。利用中间结果文件，创建一个 MappedByteBuffer, 将字符串 bytes 写入 MappedByteBuffer 中，并将 position 和 length 用位运算压缩到一个 long 值中返回。根据这个值即可从 MappedByteBuffer中读取出字符串数据。

一个优化是：如果字符串的 bytes 的长度小于等于 7: 那么直接利用 long 里面的 7 个字节存储，剩下一个字节存长度，避免了磁盘写入。附上StringStore类的核心代码：

// 写入byte[]类型的数据，范围long值作为读取的索引
public long put(byte[] data, int len) {
    if (len <= 7) {
        long value = unsafe.getLong(data, BYTE_ARRAY_BASE_OFFSET);
        return uint64(len) << 56 | (value & 0xffffffffffffffL);
    } else {
        long pos = doPut(data, len);
        return 0xffL << 56 | pos << 32 | len;
    }
}

// 根据写入时获得的long值，读取相应的数据
// 为了减少内存中的对象拷贝，直接将结果写入ByteBuffer中
public static void get(long value, ByteBuffer buf) {
    int h = (int) (value >>> 56);
    if (h != 0xff) {
        buf.putLong(value);
        buf.position(buf.position() - 8 + h);
    } else {
        int pos = (int) ((value & 0xffffff00000000L) >>> 32);
        int len = (int) (value & 0xffffffffL);
        doGet(pos, len, buf);
    }
}

数组池

如果为每行数据都创建一个 long[]，需要频繁地 new 出大量对象。为此，我们实现了一个 LongArrayPool 来管理所有的行数据，用 offset 来查找所需的数据。

对象池 + 数组化

以 Segment 中附加的 tasks 为例，如果每次 new Task() 将产生总计近亿个 Task 对象，造成严重的 GC 压力，显然不可取。

使用对象池可以解决一半的问题。通过复用 Task，减轻了 new Task() 的压力。但这还不够好！让我们看看 Task 的结构：

final class Task {
    byte opcode;
    long key;
    int promise;
    int data;
}

可见，Task 本身结构很简单，相比之下对象头的代价显得很浪费。如果不用对象，其实可以用数组来代替：

// In Segment.ensureAllocated()
opcodes = new byte[Constants.SEGMENT_MAX_EVENTS_NUM_PER_BUCKET];
offsets = new int[Constants.SEGMENT_MAX_EVENTS_NUM_PER_BUCKET];
keys = new long[Constants.SEGMENT_MAX_EVENTS_NUM_PER_BUCKET];
promises = new int[Constants.SEGMENT_MAX_EVENTS_NUM_PER_BUCKET];

这样做有以下几个优点：

• 创建（分配内存）速度大大提升
• 内存占用大幅下降，省掉了对象头的开销
• 确保连续的内存分配，顺序访问更快

对于 Promise 可以做类似的优化，参考 PromisePool，这里不再赘述。

Pool 的懒初始化

做了上述数组池和对象池优化后，程序启动时间大大增加，这是因为创建 Pool 需要大量分配内存，如果发生在类加载期间，就会阻塞 main 函数的运行。

解决方案是适当延迟部分 Pool 的分配，对它们采用 lazy 的初始化策略，即第一次使用时才分配所需的内存空间。

GC 调优

JVM 会在新生代不够分配时触发 GC。考虑到我们有 1G 的新生代内存，而事实上要动态 new 的对象很少，通过调节 Pool 的初始化时机，可以做到只发生一次 ParNew GC。

对于老年代的 CMS GC 代价很大，我们在比赛中尽可能避免触发 CMS GC。而这就要求尽可能节约内存，上文提到的对象池和数组池发挥了重要作用。

线程数调优

根据比赛数据集选取最合适的 Parser 和 Worker 线程数，对榨干最后一点 CPU 性能至关重要。

为了调优, 我们启动一个 monitor 的 daemon 线程, 定时打印 Reader, Parser 和 Worker的进度，从而推测性能瓶颈在哪一方。经尝试，我们将 Parser 线程定为 6 个，Worker（即 Bucket 数）定为 10 个。

Parser 读取预测

由于题目规定为单库单表，"变更类型"（U/I/D）之前的字符（下称 header）没有必要解析，可直接跳过。不过这部分的字符长度并不确定, 因此我们尝试预测这个跳过的长度。对于每一行，如果推测正确，则可以直接跳过这部分字符；否则，从行首开始解析直到到达"变更类型", 同时更新预估的 header 长度。

由于大部分的 log 的 header 长度是一样的, 这个技巧有效地避免大量不必要字符的解析。

同理，对于 Parser 来说除了主键以外剩下 key-value 并没有用。用类似的思路也可以预测长度并直接跳过。附上 skipHeader的代码：

if (buffer[pos + opcodeOffset - 1] == (byte) '|' 
     && buffer[pos + opcodeOffset + 1] == (byte) '|') {
    // Fast pass
    pos += opcodeOffset;
} else {
    int lineBegin = pos;
    pos++;  // Skip '|'
    pos = skipNext(buffer, pos, '|');  // Skip binlog ID
    pos = skipNext(buffer, pos, '|');  // Skip timestamp
    pos = skipNext(buffer, pos, '|');  // Skip database name
    pos = skipNext(buffer, pos, '|');  // Skip table name
    opcodeOffset = pos - lineBegin;
}      

重写网络传输和 Logging

实验发现 Netty 和 logback 都比较重量级，拖慢了启动速度。因此自己实现了网络传输和 Logger，减少启动时间。

总结与感想

算法决定了性能的上限，工程实现的好坏决定了能多大程度接近这个上限。

大赛过程竞争非常激烈，可谓高手云集。通过这场比赛，我们的技术得到了锻炼，收获了解决问题的成就感。同时也真诚感谢大赛的主办方，让我们有机会在赛场上证明自己的能力。

 

2018年阿里中间件性能挑战赛--个人总结

1.比赛结果

 

2.比赛整体感觉

初赛的题目是写一个代理实现高性能的协议解析和转发，复赛是是实现一个单机的100g的mq存储。

题目都和io有关，自己实际在比赛过程中，实现了一套能跑的方案后，后续优化的效果不明显，例如

初赛第一版6140qps，后续只优化到6470qps。

复赛第一版67w，后续只优化到115w。

在初期我的排名还能进前20，后面就迅速被n多人超越。

其实关键效果的提升，都是以好的方案为基础，调参和细节优化，提升的并不明显，所以当落后第1梯队的时候，肯定是方案有瑕疵，所以竞赛的关键

是方案优化，针对痛点的优化，不要过于依赖随机的测试。要学会在比赛中学习，学习用到比赛，提高成绩。

第1版方案很重要，其实如果一开始差距太大，对于后期自己能不能赶上，是有自我效能的降低。

对于自己的技术水平过于自信，对于别人的水平预判过低，思想上就没有重视对手。

 

3.比赛欠缺的东西

关键有效方案的优化，拿复赛举例子，如果把写缓冲和lru做了的话，预期应该能上180w。

当时没做，是用mmap方式发现加大buffer对于效果无明显提升，事后想想其实是mmap导致，如果用pwrite等系统api追加

效果应该会有提升，所以好的方案应该是符合常理，如果不符合常理应该是自己用的有问题。

把性能追求极致，之前工作总感觉用性价比最高的方式实现是有好处的，其实这样不适合比赛，因为比赛是以成绩为导向，只有把

性能追求到极致，才能竞争过别人，这个追求极致不是把n多细节优化到极致，而是大的方案基本对路，关键点做到位，其实进20不难。

不能因为方案复杂就放弃，可以考虑用开源等等。

珍惜好时间，尤其是复赛的时间，非常宝贵。

 

4.比赛好的方面

工程实现方面，复赛基本实现了bugfree，发现小问题，也基本能很快解决

把一个不太会的东西，通过测试，上线，整个流程还是很快，比如mmap的应用

复赛熬了3天到晚上2点，战斗意志很高。

 

C++ Web应用服务器中间件 MYCP 全面介绍

ESB 系列之中间件技术入门教程

1. 消息中间件
   消息中间件适用与进行网络通讯的系统，建立网络通讯的通道，进行数据和文件的传送
   产品：ActiveMQ、ZeroMQ、RabbitMQ、IBM webSphere MQ…
2. 交易中间件
   交易中间件管理分布与不同操作系统的数据，实现数据一致性，保证系统的负载均衡
   产品：IBM CICS,Bea tuxedo…
3. 对象中间件
   保证不同厂家的软件之间的交互访问
   产品：IBM componentbroker, iona orbix,borland visibroker…
4. 应用服务器
   用来构造 internet/intranet 应用和其它分布式构件应用
   产品：IBM Websphere,Bea weblogic…
5. 安全中间件
   以公钥基础设施（pki）为核心的、建立在一系列相关国际安全标准之上的一个开放式应用开发平台
   产品：entrust entrust…
6. 应用集成服务器
   把工作流和应用开发技术如消息及分布式构件结合在一起，使处理能方便自动地和构件、script 应用、工作流行为结合在一起，同时集成文档和电子邮件
   产品：lss flowman、ibm flowmark、vitria businessagiliti

##ESB##
ESB，即企业服务总线
松散耦合一直是企业软件开发中的一个很重要的内容，而面向服务的 SOA 编程在随着 ESB 的应用得到了进一步的发展，ESB 就像服务提供者和服务使用者之间的中间层

##JMS##
JMS，即 Java Message Service
ESB 仅仅是作为一个中间层，所以应用程序之间的消息通讯必须借助 JMS，即通过 JMS 从服务使用者接收消息，并将其转发到相应的服务提供者。
而且，JMS 还定义了可发送的若干不同类型的消息。例如，Text 消息包含消息的字符串表示形式；Object 消息包含序列化的 Java 对象；Map 消息包含键 / 值对的映射，等等。

附录：
MQ DEMO：

package com.wms.batchMsg;

import java.io.File;
import java.io.IOException;
import java.sql.Timestamp;
import java.text.ParseException;
import java.util.Date;

import org.apache.log4j.Logger;

import com.ibm.mq.MQEnvironment;
import com.ibm.mq.MQException;
import com.ibm.mq.MQGetMessageOptions;
import com.ibm.mq.MQMessage;
import com.ibm.mq.MQPutMessageOptions;
import com.ibm.mq.MQQueue;
import com.ibm.mq.MQQueueManager;
import com.ibm.mq.constants.MQConstants;

public class MQUtil {
	
	private static String qmName;      
    private static MQQueueManager qMgr;
    
    private static Logger logger = Logger.getLogger(MQUtil.class);
    
    static{
    	try{
	        MQEnvironment.hostname=ConfigManager.getValue("MQ_MQHost");
	        MQEnvironment.channel=ConfigManager.getValue("MQ_Server_Channel");
	        MQEnvironment.CCSID=Integer.parseInt(ConfigManager.getValue("MQ_CCSID"));
	        MQEnvironment.port=Integer.parseInt(ConfigManager.getValue("MQ_port"));
	        //MQEnvironment.userID = ConfigManager.getValue("MQ_UserId");
	        //MQEnvironment.password = ConfigManager.getValue("MQ_pass");
	        qmName = ConfigManager.getValue("MQ_QMname");
	        MQEnvironment.properties.put(MQConstants.TRANSPORT_PROPERTY,MQConstants.TRANSPORT_MQSERIES_CLIENT);
	        qMgr = new MQQueueManager(qmName);
		}catch(MQException e){
			e.printStackTrace();
			logger.info("qManager failed: Completion code " + e.completionCode + " Reason Code is "
					+ e.reasonCode);
		}
    }
    
    public static MQQueue getSendQueue(String queueName) {
		MQQueue sQueue;
        int openSendOptions = MQConstants.MQOO_OUTPUT | MQConstants.MQOO_FAIL_IF_QUIESCING
                | MQConstants.MQOO_SET_IDENTITY_CONTEXT;
        try {
            sQueue = qMgr.accessQueue(queueName, openSendOptions);
        } catch (MQException e) {
            e.printStackTrace();
            return null;
        }
        return sQueue;
    }
    
    public static MQQueue getReceiveQueue(String revQueueName){
		MQQueue rQueue ;
		int openRcvOptions = MQConstants.MQOO_INPUT_AS_Q_DEF | MQConstants.MQOO_FAIL_IF_QUIESCING;
		try{
			rQueue = qMgr.accessQueue(revQueueName, openRcvOptions);
		}catch(MQException e){
			e.printStackTrace();
			return null;
		}
		return rQueue;
	}
    
    public static void sendMsg(MQMsgEntity entity,String queueName) {
		MQQueue sendQ = null;
        try {
            MQMessage qMsg = new MQMessage();
            byte[] qByte = entity.getMsgStr().getBytes("UTF-8");
//            String message = entity.getMsgStr();
            qMsg.messageId = MQConstants.MQMI_NONE;
            //TODO send and receive
            if(entity.getCorrelId()!=null){
            	qMsg.correlationId = entity.getCorrelId();
            }
            qMsg.format = MQConstants.MQFMT_STRING;
            qMsg.write(qByte);
            MQPutMessageOptions pmo = new MQPutMessageOptions();
            pmo.options = pmo.options + MQConstants.MQPMO_NEW_MSG_ID;
            pmo.options = pmo.options + MQConstants.MQPMO_NO_SYNCPOINT;
            pmo.options = pmo.options + MQConstants.MQPMO_SET_IDENTITY_CONTEXT;
            sendQ = getSendQueue(queueName);
            sendQ.put(qMsg, pmo);
            qMgr.commit();
            //logger.info("The send message is: " +new String(qByte,"UTF-8"));
        } catch (MQException e) {
            logger.info("A WebSphere MQ error occurred : Completion code "
                            + e.completionCode + " Reason Code is "
                            + e.reasonCode);
        } catch (java.io.IOException e) {
        	logger.info("An error occurred whilst to the message buffer "
                            + e);
        }finally{
        	try{
        		if(sendQ!=null){
        			sendQ.close();
        		}
			}catch(MQException e){
				// TODO Auto-generated catch block
				e.printStackTrace();
				logger.info("Error for MQ connection:"+e.getMessage());
			}
        }

    }
    
//    public static void messageHandlerByQueueName(MQMsgEntity entity,String queueName) {
//    	try {
//    		if(queueName.equalsIgnoreCase("sap_OrdersQueue")){
//        		ECOrder order = new ECOrder();
//        		order.CallOrderCURFC(entity, "ZECI001");
//        	}else if(queueName.equalsIgnoreCase("sap_OrderPendReqQueue")){
//        		ECOrderPending orderPending = new ECOrderPending();
//        		orderPending.CallOrderPendRFC(entity, "ZECI005");
//        	}else if(queueName.equalsIgnoreCase("sap_OrderPendCancelQueue")){
//        		ECOrderPending orderPending = new ECOrderPending();
//        		orderPending.CallCancelOrderPendRFC(entity, "ZECI006");
//        	}else if(queueName.equalsIgnoreCase("sap_ECReturnsQueue")){
//        		ECOrder order = new ECOrder();
//        		order.callOrderCancelRFC(entity, "ZECI001");
//        	}else if(queueName.equalsIgnoreCase("sap_downpaymentQueue")){
//        		ECDownPayment downPayment = new ECDownPayment();
//        		downPayment.callDownPaymentRFC(entity, "ZECI007");
//        	}else if(queueName.equalsIgnoreCase("sap_360LBPQueue")){
//        		EC360LBP lbp = new EC360LBP();
//        		lbp.generateHtmlFromQueue(entity.getMsgStr());
//        	}
//		} catch (Exception e) {
//			e.printStackTrace();
//			logger.error(e.getMessage());
//		}
//    	
//    }
    
    public MQQueueManager generateNewMQQM(){
		MQQueueManager qMgr = null;
		try{
			  
	        MQEnvironment.hostname=ConfigManager.getValue("MQ_MQHost");
	        MQEnvironment.channel=ConfigManager.getValue("MQ_Server_Channel");
	        MQEnvironment.CCSID=Integer.parseInt(ConfigManager.getValue("MQ_CCSID"));
	        MQEnvironment.port=Integer.parseInt(ConfigManager.getValue("MQ_port"));   
	        String qmName = ConfigManager.getValue("MQ_QMname");
	        MQEnvironment.properties.put(MQConstants.TRANSPORT_PROPERTY,MQConstants.TRANSPORT_MQSERIES_CLIENT);
	        qMgr = new MQQueueManager(qmName);
	        
		}catch(MQException e){
			e.printStackTrace();
			logger.info("qManager failed: Completion code " + e.completionCode + " Reason Code is "
					+ e.reasonCode);
		}
		return qMgr;
	}
    
    public void MultiThreadGetMqMessage(MQQueueManager qMgr,String queueName){
    	MQQueue revQ = null;
        String mqString = null;
        MQMsgEntity entity = new MQMsgEntity();
        
        int openRcvOptions = MQConstants.MQOO_INPUT_AS_Q_DEF | MQConstants.MQOO_FAIL_IF_QUIESCING;
        try {
        	MQMessage retrievedMessage = new MQMessage();
            MQGetMessageOptions gmo = new MQGetMessageOptions();
            gmo.options += MQConstants.MQPMO_NO_SYNCPOINT;// 
            gmo.options = gmo.options + MQConstants.MQGMO_WAIT;//
            gmo.options = gmo.options + MQConstants.MQGMO_FAIL_IF_QUIESCING;// 
            gmo.waitInterval = MQConstants.MQWI_UNLIMITED;// 
            gmo.matchOptions = MQConstants.MQMO_MATCH_MSG_ID;
            retrievedMessage.format=MQConstants.MQFMT_STRING;
            //  MQC.MQWI_UNLIMITED;
            revQ = qMgr.accessQueue(queueName, openRcvOptions);
            revQ.get(retrievedMessage, gmo);
            qMgr.commit();
            int length = retrievedMessage.getDataLength();
            if(length >0){
            	long startTime = System.currentTimeMillis();
	            byte[] msg = new byte[length];
	            retrievedMessage.readFully(msg);
	            mqString = new String(msg, "UTF-8");
	            if(queueName.equalsIgnoreCase("sap_360LBPQueue")){
	            	mqString = mqString.replace("''", "\"");
	            }
	            long timeuse = System.currentTimeMillis() - startTime;
	            Date currentDate = new Date();
    			Timestamp receiveTimestamp = new Timestamp(currentDate.getTime());
	            logger.info("=========mqString from "+queueName+" :"+mqString);
	            DBUtil.insertIntoMQLog("Receive",queueName, mqString, timeuse, "success", "", null, receiveTimestamp);
	            entity.setMsgStr(mqString);
	            //messageHandlerByQueueName(entity,queueName);
	            
    			
            }else{
            	logger.info("Error MQ string Sent!");
            }
        }
        catch (MQException e) {
        	e.printStackTrace();
            if (e.reasonCode != 2033) 
            {
            	logger.info(e.getMessage());
                logger.info("Completion code "
                        + e.completionCode + " Reason Code is " + e.reasonCode);
            }
        } catch (IOException e) {
            logger.info("IO error:" + e.getMessage());
        } finally{
        	try{
        		if(revQ!=null){
        			revQ.close();
        		}
			}catch(MQException mqEx){
				int rc = mqEx.reasonCode;
				if (rc != MQException.MQRC_NO_MSG_AVAILABLE)
				{
					logger.info(" PUT Message failed with rc = " 
				+ rc);
				}

			}
        }
    }
    
    public static String getMQMessage(String queueName) throws ParseException {
        MQQueue revQ = null;
        String mqString = null;
        MQMsgEntity entity = new MQMsgEntity();
        try {
        	MQMessage retrievedMessage = new MQMessage();
            MQGetMessageOptions gmo = new MQGetMessageOptions();
            gmo.options += MQConstants.MQPMO_NO_SYNCPOINT;// 
            gmo.options = gmo.options + MQConstants.MQGMO_WAIT;//
            gmo.options = gmo.options + MQConstants.MQGMO_FAIL_IF_QUIESCING;// 
            gmo.waitInterval = MQConstants.MQWI_UNLIMITED;// 
            gmo.matchOptions = MQConstants.MQMO_MATCH_MSG_ID;
            retrievedMessage.format=MQConstants.MQFMT_STRING;
            //  MQC.MQWI_UNLIMITED;
            revQ = getReceiveQueue(queueName);
            revQ.get(retrievedMessage, gmo);
            qMgr.commit();
            int length = retrievedMessage.getDataLength();
            if(length >0){
	            byte[] msg = new byte[length];
	            retrievedMessage.readFully(msg);
	            mqString = new String(msg, "UTF-8");
	            logger.info("=========getMQMessage===mqString from "+queueName+" :"+mqString);
	            entity.setMsgStr(mqString);
	            //messageHandlerByQueueName(entity,queueName);
            }else{
            	logger.info("Error MQ string Sent!");
            }
        }
        catch (MQException e) {
        	e.printStackTrace();
            if (e.reasonCode != 2033) 
            {
                e.printStackTrace();
                logger.info("Completion code "
                        + e.completionCode + " Reason Code is " + e.reasonCode);
            }
        } catch (java.io.IOException e) {
            System.out.println("error" + e.getMessage());
        }finally{
        	try{
        		if(revQ!=null){
        			revQ.close();
        		}
			}catch(MQException mqEx){
				int rc = mqEx.reasonCode;
				if (rc != MQException.MQRC_NO_MSG_AVAILABLE)
				{
				System.out.println(" PUT Message failed with rc = " 
				+ rc);
				}

			}
        }
        return mqString;
    }

	public void revAndSend(MQMsgEntity entity,String queueName){
		//
		sendMsg(entity,queueName);
	}
	
	public void subscribeMessage() throws ParseException{
		while(true){
			logger.info("waiting to get message.....");
			getMQMessage("sap_OrdersQueue");
			
		}
	}
	
	public void subscribeOrderPendMessage() throws ParseException{
		while(true){
			logger.info("waiting to get message.....");
			getMQMessage("sap_ECReturnsQueue");
			
		}
	}
	
	
	
	public  static void main(String[] args) throws IOException, ParseException {
		MQMsgEntity entity = new MQMsgEntity();
		String sendMsg = XMLBeanUtil.readFileToString(new File("D://batchXML0108.txt"));
		int intPktCtlNbr = 1;
		String StrPkt = null;
		String newPktCtlNbr =null;
		for (int i = 0; i < 20000; i++) {
			newPktCtlNbr = String.format("%09d", intPktCtlNbr+i);  
			StrPkt="<PktCtlNbr>"+"V"+newPktCtlNbr+"</PktCtlNbr>";
			String changeSendMsg = sendMsg.replaceAll("<PktCtlNbr>6001996171</PktCtlNbr>", StrPkt);
			System.out.println(StrPkt);
			entity.setMsgStr(changeSendMsg);
			sendMsg(entity,"wms_SAPOrderQueue");
		}
//		MQUtil util = new MQUtil();
//		util.subscribeMessage();
//		util.subscribeOrderPendMessage();
//		util.messageHandlerByQueueName(entity, "sap_360LBPQueue");
//		getMQMessage("sap_OrderPendCancelQueue");
//		System.out.println("rev message is:"+message);
	}

}

本文分享 CSDN - smileNicky。
如有侵权，请联系 support@oschina.cn 删除。
本文参与 “OSC 源创计划”，欢迎正在阅读的你也加入，一起分享。

关于阿里中间件技术：应用服务器篇和阿里 中间件的介绍已经告一段落，感谢您的耐心阅读，如果想了解更多关于10G mysql binlog重放并传输到另一台服务器执行，阿里中间件大赛、2018年阿里中间件性能挑战赛--个人总结、C++ Web应用服务器中间件 MYCP 全面介绍、ESB 系列之中间件技术入门教程的相关信息，请在本站寻找。

本文标签：

• 阿里中间件技术
• 应用服务器篇
• 阿里巴巴
• 中间件