本文由ilanniweb微信公众号提供友情赞助，首发于烂泥行天下

jenkins技术分享QQ群：571981257

前段时间写了两篇有关Activemq搭建和集群相关的文章，文章链接为《烂泥：ActiveMQ的集群安装与配置》。

但是由于Activemq集群的特殊性，经常出现Activemq集群假死的情况。为了保证集群的正常运行，就需要我们监控集群的运行情况。

要监控Activemq集群的运行情况，我们就可以通过stomp协议的相关客户端来实现。

一、stomp.py简介

在此我们使用的是stomp协议的python客户端，来实现集群的监控。

支持stomp协议的python客户端，我们使用的是stomp.py这个软件来实现，而stomp.py我们可以在github上来获得。

stomp.py除了可以连接activemq，还可以连接rabbitmq。

stomp.py仓库的github地址为：

https://github.com/jasonrbriggs/stomp.py.git

考虑到从github仓库拉取速度的情况，我现在在码云上同步了该仓库，地址如下：

https://gitee.com/ilanni/stomp.py.git

二、初始化环境

在正式配置监控Activemq集群脚本之前，我们先来初始化stomp.py运行的环境，如下：

wget -O /etc/yum.repos.d/CentOS-Base.repo http://mirrors.163.com/.help/CentOS7-Base-163.repo

rpm -ivh http://mirrors.ustc.edu.cn/epel/7/x86_64/e/epel-release-7-10.noarch.rpm

安装stomp.py运行所需的python模块，如下：

yum -y install python-paramiko

到此环境初始化完毕。

三、下载并安装stomp.py

环境初始化完毕后，我们现在来安装和配置stomp.py。如下：

git clone https://gitee.com/ilanni/stomp.py.git

cd stomp.py

python setup.py install

四、集群监控脚本

Acitvemq集群的监控脚本，如下：

vim ilanni.py

#!/usr/bin/env python

#coding=utf-8

import paramiko

import time

import sys

import stomp

import datetime

time_stamp = datetime.datetime.now()

Aip = (“10.1.1.127”)

Bip = (“10.1.1.128”)

Cip = (“10.1.1.129″)

list = []

def Checknetstat():

for i in (Aip,Bip,Cip):

hostname=i

username=”root”

password=”123456″

paramiko.util.log_to_file(“syslogin.log”)

ssh=paramiko.SSHClient()

ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

ssh.load_system_host_keys()

ssh.connect(hostname=hostname,username=username,password=password,allow_agent=False,look_for_keys=False)

stdin,stdout,stderr=ssh.exec_command(“netstat -ntpl | grep 61616 | wc -l”)

word=str(stdout.read().strip(‘\n’))

list.append(word)

ssh.close()

print list

def GoListen(xip):

class MyListener(object):

def on_error(self, headers, message):

print(‘received an error %s’ % message)

def on_message(self, headers, message):

print(‘received a message %s’ % message)

conn = stomp.Connection([(xip,61613)])

conn.set_listener(”, MyListener())

conn.start()

conn.connect(‘admin’,’admin’)

conn.subscribe(destination=’/queue/test_qjd_mq’, id=1, ack=’auto’)

conn.send(body=’qjd test activemq cluster’, destination=’/queue/test_qjd_mq’)

time.sleep(2)

conn.disconnect()

def Restart():

for i in (Aip,Bip,Cip):

hostname=i

username=”root”

password=”123456″

paramiko.util.log_to_file(“syslogin.log”)

ssh=paramiko.SSHClient()

ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

ssh.load_system_host_keys()

ssh.connect(hostname=hostname,username=username,password=password,allow_agent=False,look_for_keys=False)

stdin,stdout,stderr=ssh.exec_command(“ps -ef | grep activemq| grep -v grep | awk ‘{print $2}’ |xargs kill -9;export JAVA_HOME=/usr/local/jdk1.8.0_144;/usr/local/activemq/bin/activemq start“)

print stdout.read()

ssh.close()

def Remotecommand():

if list == [‘1’, ‘0’, ‘0’]:

print “time_stamp ” + time_stamp.strftime(‘%Y.%m.%d-%H:%M:%S’)

print (“第一台机器 %s 的61616端口处于监听状态，现在尝试往里面发送消息…”) % Aip

try:

GoListen(Aip)

except stomp.exception.ConnectFailedException:

print (“发送信息失败，请重启Activemq进程…”)

elif list == [‘0’, ‘1’, ‘0’]:

print “time_stamp ” + time_stamp.strftime(‘%Y.%m.%d-%H:%M:%S’)

print (“第二台机器 %s 的61616端口处于监听状态，现在尝试往里面发送消息…”) % Bip

try:

GoListen(Bip)

except stomp.exception.ConnectFailedException:

print (“发送信息失败，请重启Activemq进程…”)

elif list == [‘0’, ‘0’, ‘1’]:

print “time_stamp ” + time_stamp.strftime(‘%Y.%m.%d-%H:%M:%S’)

print (“第三台机器 %s 的61616端口处于监听状态，现在尝试往里面发送消息…”) % Cip

try:

GoLinten(Cip)

except stomp.exception.ConnectFailedException:

print (“发送信息失败，请重启Activemq进程…”)

else:

print “time_stamp ” + time_stamp.strftime(‘%Y.%m.%d-%H:%M:%S’)

print (“activemq集群异常，现在重启这三台mq…”)

Restart()

if __name__ == “__main__”:

Checknetstat()

Remotecommand()

print(“整个脚本执行结束，感谢您的使用！”)

该脚本的整体思路是，通过该脚本向Activemq集群发送消息。如果向集群发送消息不成功(排除网络和脚本问题)，则说明集群出现问题。

判断集群出现问题后，则通过该脚本登陆到Activemq集群机器上kill掉activemq相关的进程，重启Activemq服务即可。

该脚本需要注意的地方：

1、脚本中连接Activemq集群机器是通过用户名和密码，如下：

2、脚本中启动Activemq服务要导入jdk环境变量，如下：

3、Activemq集群web管理平台的用户名和密码均为admin，不过这个是可以修改的。如下：

4、该脚本中向Activemq集群中产生的队列名称为test_qjd_mq，如下：

脚本可以直接下载：点我下载。

该脚本我们使用crontab，让其定时运行。如下：

vim /etc/crontab

*/2 * * * * root /usr/bin/python /usr/local/stomp/ilanni.py >>/tmp/mq.log

五、访问Acitvemq集群web管理页面

在Acitvemq集群里面，此时我们不知道哪个节点是master状态，所以我们也无法通过activemq的web管理页面查看activemq集群的信息。

其实对于这种情况，我们完全可以通过nginx的upstream来访问activemq集群的web管理页面，如下：

通过上图，我们可以看到activemq集群运行正常。

现在我们来验证下，上述监控脚本是否是正常运行的。

六、验证集群监控脚本

现在我们要通过activemq的web管理页面，验证Activemq集群监控脚本。

打开activemq集群的web管理页面，如下：

通过上图，我们可以很明显的看到test_qjd_mq队列中有3270个消息等待消费，这3270个消息就是我们的监控脚本产生的

这就说明集群监控脚本，已经成功向activemq集群生产消息。

现在我们手工停掉三台机器上的activemq服务，如下：

ps -ef | grep activemq| grep -v grep | awk ‘{print $2}’ |xargs kill -9

ps -ef | grep activemq

2分钟以后，我们登陆web管理页面，如下：

通过上图，我们可以很明显的看到现在test_qjd_mq队列中只有4条消息未被消费掉了，这就说明了监控脚本会在监测到activemq服务未启动时，自动启动三台机器上的activemq服务。

通过上图测试，说明我们的集群监控脚本是有效的。

未经允许不得转载：烂泥行天下 » 烂泥：Activemq集群监控脚本