RabbitMQ Management HTTP API的简单封装

[TOC]

官方相关

官方文档：https://rawcdn.githack.com/rabbitmq/rabbitmq-server/v3.10.0/deps/rabbitmq_management/priv/www/api/index.html

支持版本：rabbitmq-v3.10.0

介绍翻译：所有URI将仅服务于application/json类型的资源，并且将需要HTTP Basic Auth认证。许多URI要求虚拟主机的名称作为路径的一部分，因为名称只能唯一地标识虚拟主机中的对象。由于默认虚拟主机名为”/“，因此需要将其编码为”%2F”。传参的JSON对象必须具有某些必选参数，可选参数是可以被忽略的。缺少必选参数将导致请求错误。

部分文档截图：

部分文档截图

代码展示

基于Spring框架以及Hutool，Fastjson等工具依赖

配置文件bootstrap.yml，仅供参考

1
2
3
4
rms:
  username: admin
  password: admin
  managementUrl: http:///192.168.200.250:15672

配置类RmsConfig.java

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Configuration;
import lombok.Data;

/**
 * @ClassName：RmsConfig
 * @Description：配置类
 * @Author：Administrator
 * @Date：2022年5月11日下午23:17:03
 */
@Data
@Configuration
@ConfigurationProperties(prefix = "rms", ignoreInvalidFields = true)
public class RmsConfig {

	/** 管理地址 */ 
	private String managementUrl;
	/** 连接用户名 */ 
    private String username;
    /** 连接密码 */ 
    private String password;
    
}

RabbitMQ管理API工具类RabbitmqHttpApi.java，这里拿创建用户和用户授权做例子，其他接口可以举一反三实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.DependsOn;
import org.springframework.stereotype.Component;
import com.alibaba.fastjson.JSONPath;
import cn.hutool.core.convert.Convert;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpRequest;
import cn.hutool.http.HttpResponse;
import cn.hutool.http.HttpStatus;

/**
 * @ClassName：RabbitmqHttpApi
 * @Description：RabbitMQ管理API工具类
 * @Author：Administrator
 * @Date：2022年5月11日下午23:23:33
 */
@DependsOn("rmsConfig")
@Component
public class RabbitmqHttpApi {

	private static String managementUrl;
	private static String adminUsername;
	private static String adminPassword;

	@Autowired
	public void setMqttConfig(RmsConfig rmsConfig) {
		RabbitmqHttpApi.managementUrl = rmsConfig.getManagementUrl();
		RabbitmqHttpApi.adminUsername = rmsConfig.getUsername();
		RabbitmqHttpApi.adminPassword = rmsConfig.getPassword();
	}

	/**
	 * @ClassName：Tags
	 * @Description：rabbitmq角色tag
	 * @Author：Administrator
	 * @Date：2022年5月11日下午23:25:04
	 */
	public enum Tags {
		/** 超级管理员，可登陆管理控制台，可查看所有的信息，并且可以对用户，策略(policy)进行操作 */
		ADMINISTRATOR("administrator"),
		/** 监控者，可登陆管理控制台，同时可以查看rabbitmq节点的相关信息(进程数，内存使用情况，磁盘使用情况等) */
		MONITORING("monitoring"),
		/** 策略制定者，可登陆管理控制台，同时可以对policy进行管理 */
		POLICYMAKER("policymaker"),
		/** 普通管理者，仅可登陆管理控制台，无法看到节点信息，也无法对策略进行管理 */
		MANAGEMENT("management"),
		/** 模拟者，无法登录管理控制台 */
		IMPERSONATOR("impersonator"),
		/** 其他，无法登陆管理控制台，通常就是普通的生产者和消费者 */
		NONE("");
		private final String text;
		private Tags(String text) {
			this.text = text;
		}
	}

	/**
	 * 创建普通用户
	 * @param username
	 * @param password
	 */
	public static boolean createCommonUser(String username, String password) throws Exception {
		return createUser(username, password, Tags.NONE);
	}

	/**
	 * 创建超级用户
	 * @param username
	 * @param password
	 */
	public static boolean createAdminUser(String username, String password) throws Exception {
		return createUser(username, password, Tags.ADMINISTRATOR);
	}

	/**
	 * 创建用户
	 * 参考文档`/api/users/{name}`接口
	 * @param username
	 * @param password
	 */
	public static boolean createUser(String username, String password, Tags tags) throws Exception {
		String param = String.format("{\"password\":\"%s\",\"tags\":\"%s\"}", password, tags.text);
		String url = String.format("%s/api/users/%s", managementUrl, username);
		HttpResponse response = HttpRequest.put(url).basicAuth(adminUsername, adminPassword).body(param).execute();
		return response.getStatus() == HttpStatus.HTTP_CREATED || response.getStatus() == HttpStatus.HTTP_NO_CONTENT;
	}

	/**
	 * 设置所有权限
	 * @param username
	 */
	public static boolean setAllPermissions(String username) throws Exception {
		return setPermissions(username, "/", ".*", ".*", ".*");
	}

	/**
	 * 设置权限
	 * 参考文档`/api/permissions/{vhost}/{name}`接口
	 * @param username
	 * @param virtualHost
	 * @param configureRegexp
	 * @param writeRegexp
	 * @param readRegexp
	 */
	public static boolean setPermissions(String username, String virtualHost, String configureRegexp,
			String writeRegexp, String readRegexp) throws Exception {
		String param = String.format("{\"configure\":\".*\",\"write\":\".*\",\"read\":\".*\"}", configureRegexp,
				writeRegexp, readRegexp);
		virtualHost = StrUtil.equals("/", virtualHost) ? "%2F" : virtualHost;
		String url = managementUrl + "/api/permissions/" + virtualHost + "/" + username;
		HttpResponse response = HttpRequest.put(url).basicAuth(adminUsername, adminPassword).body(param).execute();
		return response.getStatus() == HttpStatus.HTTP_CREATED || response.getStatus() == HttpStatus.HTTP_NO_CONTENT;
	}
    
    /**
	 * 获取集群节点socket使用数量
	 * 参考文档`/api/nodes`接口
	 * @return array
	 */
	public static Integer[] getUsedSockets() throws Exception {
		String url = managementUrl + "/api/nodes";
		HttpResponse response = HttpRequest.get(url).basicAuth(adminUsername, adminPassword).execute();
		Object object = JSONPath.read(response.body(), "$['sockets_used']");
		return ObjectUtil.isNotNull(object) ? Convert.toIntArray(object) : new Integer[] {};
	}

}

在业务中调用，仅供参考

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
/**
 * 某某业务实现
 * @param username
 * @param password
 */
public void doBiz(String username, String password) {
    if (createClientUser(username, password)) {
        logger.info("创建rabbitmq用户【{}】成功", username);
    }
    Integer nodeIndex = dispatchNode();
    if (!ObjectUtil.isNull(nodeIndex)) {
        logger.info("节点【{}】连接数最少，进行调度", nodeIndex + 1);
    }
}

/**
 * 创建连接客户端用户
 * @param username
 * @param password
 */
public boolean createClientUser(String username, String password) {
    try {
        return RabbitmqHttpApi.createCommonUser(username, password) && RabbitmqHttpApi.setAllPermissions(username);
    } catch (Exception e) {
        logger.error("创建rabbitmq用户【{}】异常：{}", username, e.getMessage());
        return false;
    }
}

/** 调度mqtt节点 */
public Integer dispatchNode() {
    try {
        Integer[] useds = RabbitmqHttpApi.getUsedSockets();
        if (useds.length == 0) {
            throw new Exception("接口未获取到集群各节点socket使用数量");
        }
        return ArrayUtil.indexOf(useds, NumberUtil.min(useds));
    } catch (Exception e) {
        logger.error("调度mqtt节点异常：{}", e.getMessage());
    }
    return null;
}