部署IBM MQ集群

IT那活儿 2024-11-18

255

点击上方“IT那活儿”公众号--专注于企业全栈运维技术分享，不管IT什么活儿，干就完了！！！

问

为什么要使用集群？

集群可以在集群中的队列和队列管理器之间进行消息分配来达到消息负载均衡的目的，可以提升系统的容灾能力以及提高系统的性能。

只要这些货物在集装箱里封装好，就可以用一艘大船把他们都运走。

答

一

集群的组成

1.1 集群存储库

集群存储库是集群成员的队列管理器的集合。

存储库所包含的信息有：

队列管理器的名称、队列管理器的位置、该队列管理器托管的队列、该队列管理器的通道以及其他信息。

集群存储库又分为完全存储库和部分存储库：

完全存储库拥有集群中每个队列管理器完整信息；
部分群存储库仅包含当前队列管理器和与当前队列管理器有信息交换的队列管理器的信息。

为确保在集群队列管理器不可用时不会丢失消息，一个集群应最少两个队列管理器拥有完全存储库。

1.2 集群队列管理器

集群中的队列管理器叫集群管理器，一个队列管理器可以是多个集群的成员，但是他在所属的集群中必须拥有唯一的名称。

1.3 集群队列

集群队列是由集群托管的可供集群中其他队列管理器使用队列。集群中的所有队列管理器都可以向该队列发送消息，但只有在定义了该队列的集群管理器中可以取出消息。

1.4 集群通道

集群通道分为集群接收通道（CLUSRCVR）与集群发送通道（CLUSSDR）。

集群中的队列管理器至少应具有一个集群接收通道，通过定义集群接收通道通道，该队列管理器向其他集群队列管理器表示它可用于接收消息；
集群中的完整存储库交换的所有更新都在集群发送通道上流动，部分存储库的集群发送通道必须指向包含完全存储库的集群管理器，选择哪个完全存储库没有区别，因为在定义之后集群会根据需要进行自定义通道。

二

集群环境

IP	Usage
10.17.0.117	完全存储库CQM1
10.17.0.118	完全存储库CQM2
10.17.0.119	完全存储库CQM3
10.17.0.120	完全存储库CQM4
10.17.0.121	完全存储库CQM5

注：IBM MQ 集群5.1以上的版本。

2.1 创建集群完全存储库CQM1

#! bin/bash
QmgrName=CQM1
#创建队列管理器
crtmqm $QmgrName
#启动队列管理器
strmqm $QmgrName
#定义一个服务器连接通道（供管理工具与代码调用使用，此处可去掉）
echo "DEFINE CHANNEL(TESTCHANNEL) CHLTYPE(SVRCONN) TRPTYPE(TCP) MCAUSER('mqm')" | runmqsc $QmgrName
#修改默认监听端口为1414
echo "DEFINE LISTENER(LISTENER.TCP) TRPTYPE(TCP) PORT(1414)" | runmqsc $QmgrName
#队列管理器权限，禁用认证
echo "SET CHLAUTH(*) TYPE(BLOCKUSER) USERLIST(*MQADMIN) ACTION(REMOVE)" | runmqsc $QmgrName
echo "SET CHLAUTH(SYSTEM.ADMIN.SVRCONN) TYPE(ADDRESSMAP) ADDRESS(*) ACTION(REMOVE)" | runmqsc $QmgrName
echo "SET CHLAUTH(SYSTEM.*) TYPE(ADDRESSMAP) ADDRESS(*) ACTION(REMOVE)" | runmqsc $QmgrName
echo "ALTER AUTHINFO(SYSTEM.DEFAULT.AUTHINFO.IDPWOS) AUTHTYPE(IDPWOS) CHCKCLNT(OPTIONAL)" | runmqsc $QmgrName
#启动监听器
echo "START LISTENER(LISTENER.TCP)" | runmqsc $QmgrName
#禁用通道验证
echo "alter qmgr chlauth(disabled)" | runmqsc $QmgrName
#刷新安全策略
echo "REFRESH SECURITY TYPE(CONNAUTH)" | runmqsc $QmgrName
#设置该队列管理器为集群（MYClUSTER）的完全仓储库
echo "ALTER QMGR REPOS(MYClUSTER)" | runmqsc $QmgrName
#定义集群（MYClUSTER）接收通道
echo "DEFINE CHANNEL(TO.CQM1) CHLTYPE(CLUSRCVR) TRPTYPE(TCP) CONNAME('10.17.0.117(1414)') CLUSTER(MYClUSTER)" | runmqsc $QmgrName
#定义集群（MYClUSTER）发送通道
echo "DEFINE CHANNEL(TO.CQM2) CHLTYPE(CLUSSDR) TRPTYPE(TCP) CONNAME('10.17.0.117 (1414)') CLUSTER(MYClUSTER)" | runmqsc $QmgrName

2.2 创建集群完全存储库CQM2

与创建集群完全存储库CQM1脚本差不多，只是修改了队列管理器的名称和集群发送通道与集群接收通道的名称。

#! /bin/bash
QmgrName=CQM2
#创建队列管理器
crtmqm $QmgrName
#启动队列管理器
strmqm $QmgrName
#定义一个服务器连接通道
echo "DEFINE CHANNEL(TESTCHANNEL) CHLTYPE(SVRCONN) TRPTYPE(TCP) MCAUSER('mqm')" | runmqsc $QmgrName
echo "DEFINE LISTENER(LISTENER.TCP) TRPTYPE(TCP) PORT(1414)" | runmqsc $QmgrName
#队列管理器权限，禁用认证
echo "SET CHLAUTH(*) TYPE(BLOCKUSER) USERLIST(*MQADMIN) ACTION(REMOVE)" | runmqsc $QmgrName
echo "SET CHLAUTH(SYSTEM.ADMIN.SVRCONN) TYPE(ADDRESSMAP) ADDRESS(*) ACTION(REMOVE)" | runmqsc $QmgrName
echo "SET CHLAUTH(SYSTEM.*) TYPE(ADDRESSMAP) ADDRESS(*) ACTION(REMOVE)" | runmqsc $QmgrName
echo "ALTER AUTHINFO(SYSTEM.DEFAULT.AUTHINFO.IDPWOS) AUTHTYPE(IDPWOS) CHCKCLNT(OPTIONAL)" | runmqsc $QmgrName
#启动监听器
echo "START LISTENER(LISTENER.TCP)" | runmqsc $QmgrName
#禁用通道验证
echo "alter qmgr chlauth(disabled)" | runmqsc $QmgrName
#刷新安全策略
echo "REFRESH SECURITY TYPE(CONNAUTH)" | runmqsc $QmgrName
#设置该队列管理器为集群（MYClUSTER）的完全仓储库
echo "ALTER QMGR REPOS(MYClUSTER)" | runmqsc $QmgrName
#定义集群（MYClUSTER）接收通道
echo "DEFINE CHANNEL(TO.CQM2) CHLTYPE(CLUSRCVR) TRPTYPE(TCP) CONNAME('10.17.0.118(1414)') CLUSTER(MYClUSTER)" | runmqsc $QmgrName
#定义集群（MYClUSTER）发送通道
echo "DEFINE CHANNEL(TO.CQM1) CHLTYPE(CLUSSDR) TRPTYPE(TCP) CONNAME('10.17.0.117(1414)') CLUSTER(MYClUSTER)" | runmqsc $QmgrName

2.3 创建集群部分存储库CQM3，并定义集群队列CQ1

相比创建完全存储库的脚本，少了将队列管理器设置为完全存储库的命令。

注：

若集群中的多个队列管理器包含统一名称的队列，那么在集群中没有定义该队列的队列管理器上向该队列放入消息即可达到负载均衡的目的，集群会自动将消息分发到集群中所有包含该队列的队列管理器中的响应队列上，但若发送消息的队列管理器本身也定义了该队列，那么消息只会进入到当前队列管理器的队列中，不会参与负载均衡。

#! /bin/bash
QmgrName=CQM3
#创建队列管理器
crtmqm $QmgrName
#启动队列管理器
strmqm $QmgrName
#定义一个服务器连接通道
echo "DEFINE CHANNEL(TESTCHANNEL) CHLTYPE(SVRCONN) TRPTYPE(TCP) MCAUSER('mqm')" | runmqsc $QmgrName
echo "DEFINE LISTENER(LISTENER.TCP) TRPTYPE(TCP) PORT(1414)" | runmqsc $QmgrName
#队列管理器权限，禁用认证
echo "SET CHLAUTH(*) TYPE(BLOCKUSER) USERLIST(*MQADMIN) ACTION(REMOVE)" | runmqsc $QmgrName
echo "SET CHLAUTH(SYSTEM.ADMIN.SVRCONN) TYPE(ADDRESSMAP) ADDRESS(*) ACTION(REMOVE)" | runmqsc $QmgrName
echo "SET CHLAUTH(SYSTEM.*) TYPE(ADDRESSMAP) ADDRESS(*) ACTION(REMOVE)" | runmqsc $QmgrName
echo "ALTER AUTHINFO(SYSTEM.DEFAULT.AUTHINFO.IDPWOS) AUTHTYPE(IDPWOS) CHCKCLNT(OPTIONAL)" | runmqsc $QmgrName
#启动监听器
echo "START LISTENER(LISTENER.TCP)" | runmqsc $QmgrName
#禁用通道验证
echo "alter qmgr chlauth(disabled)" | runmqsc $QmgrName
#刷新安全策略
echo "REFRESH SECURITY TYPE(CONNAUTH)" | runmqsc $QmgrName
#定义集群（MYClUSTER）接收通道
echo "DEFINE CHANNEL(TO.CQM3) CHLTYPE(CLUSRCVR) TRPTYPE(TCP) CONNAME('10.17.0.119(1414)') CLUSTER(MYClUSTER)" | runmqsc $QmgrName
#定义集群（MYClUSTER）发送通道
echo "DEFINE CHANNEL(TO.CQM1) CHLTYPE(CLUSSDR) TRPTYPE(TCP) CONNAME('10.17.0.117(1414)') CLUSTER(MYClUSTER)" | runmqsc $QmgrName
#定义集群共享队列
echo "DEFINE QLOCAL(CQ1) CLUSTER(MYClUSTER)" | runmqsc $QmgrName

2.4 创建集群部分存储库CQM4、CQM5，并定义集群队列CQ1

集群部分存储库的创建都差不多，参考创建部分存储库CQM3的脚本，请自行处理。

注意队列管理器名称与集群发送通道和集群接收通道的名称即可。

2.5 集群测试

这里使用java代码向集群中写入消息，三个虚拟机使用命令进行消息接收，测试负载均衡。

package com.qinke;

import com.ibm.mq.*;
import com.ibm.mq.constants.MQConstants;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;

public class ClusterTest {
  private static String hostname = "10.17.0.117";
  private static Integer port = 1414;
  private static String channel = "TESTCHANNEL";
  private static String qmgrName = "CQM1";
  private static String queueName = "CQ1";
    private MQQueueManager queueManager = null;
    @Before
    public void init() {
        MQEnvironment.hostname = hostname ;
        MQEnvironment.port = port;
        MQEnvironment.channel = channel ;
        try {
            queueManager = new MQQueueManager(qmgrName );
        } catch (MQException e) {
            e.printStackTrace();
        }
    }

    @Test
    public void clusterSend() {
        System.out.println("Starting Sender");
        MQQueue q = null;

        try {
            q = queueManager.accessQueue(queueName , MQConstants.MQOO_OUTPUT+ MQConstants.MQOO_BIND_NOT_FIXED);
            for (int i = 1; i <= 20; i++) {
                MQMessage message = new MQMessage();
                message.writeString(" Test message " + i);
                q.put(message);
                System.out.println("Put test message " + i);
            }
            q.close();
        } catch (MQException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (Exception e) {
            e.printStackTrace();
        } catch (Throwable t) {
            t.printStackTrace();
        } finally {
            try {
                if (q != null && q.isOpen()) {
                    q.close();
                }
                if (queueManager != null && queueManager.isConnected()) {
                    queueManager.disconnect();
                }
            } catch (MQException ex) {
                ex.printStackTrace();
            }
        }
    }}

2.6 消息发送程序

注：

队列打开方式必须有 MQOO_BIND_NOT_FIXED，否则同一个连接发送的消息只会执行一次负载均衡算法，即此连接发送的所有消息都会到同一个队列管理器。

END

本文作者：赵乐乐(上海新炬中北团队）

本文来源：“IT那活儿”公众号

集群服务器集群技术

文章转载自IT那活儿，如果涉嫌侵权，请发送邮件至：contact@modb.pro进行举报，并提供相关证据，一经查实，墨天轮将立刻删除相关内容。