Fork me on GitHub

Redis的消息订阅/发布 Utils工具类

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
129
130
package cn.cicoding.utils;

import org.json.JSONException;
import org.json.JSONObject;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.JedisPubSub;
import redis.clients.jedis.Protocol;
import redis.clients.jedis.exceptions.JedisConnectionException;

class MQClient {
public static final int MSG_REALTIME = 1;
public static final int MSG_CACHED = 2;
public static final int MSG_SERIALIZABLE = 3;
public static final String NOTIFY_CHANNEL = "ClientNotify";

private JedisPool pool;
private boolean exit;
private JedisPubSub pubsub;

public MQClient(String ip, int port, JedisPubSub pubsub) {
JedisPoolConfig config = new JedisPoolConfig();
config.setMaxWaitMillis(10 * 1000);
config.setMaxIdle(1000);
config.setTestOnBorrow(true);
pool = new JedisPool(config, ip, port, Protocol.DEFAULT_TIMEOUT, null);
exit = false;
this.pubsub = pubsub;
}

public boolean publish(String channels, String message, String content) {
JSONObject obj = new JSONObject();
boolean ret = false;
Jedis jedis = null;
try {
jedis = pool.getResource();
if (message != null) {
obj.put("message", message);
}
try {
JSONObject objCon = new JSONObject(content);
obj.put("content", objCon);
} catch (JSONException e) {
obj.put("content", content);
}
String[] tmp = channels.split(";");
for (String channel : tmp) {
try {
if (jedis.publish(channel, obj.toString()) > 0) {
ret = true;
}
} catch (Exception e) {
break;
}
}
} catch (JSONException e) {
} finally {
if (jedis != null){
jedis.close();
}
}

return ret;
}

public boolean clientNotify(String clients, String message, String content, int type) {
if (type == MSG_REALTIME) {
return publish(clients, message, content);
}

boolean ret = false;

try {
JSONObject obj = new JSONObject();
obj.put("clients", clients);
obj.put("type", type);
if (message != null) {
obj.put("message", message);
}
try {
JSONObject objCon = new JSONObject(content);
obj.put("content", objCon);
} catch (JSONException e) {
obj.put("content", content);
}

if (pool.getResource().publish(NOTIFY_CHANNEL, obj.toString()) > 0) {
ret = true;
}
} catch (JSONException e) {
}

return ret;
}

public boolean setValue(String key, String value) {
try {
String response = pool.getResource().set(key, value);
if (response != null && response.equals("OK")) {
return true;
}
} catch (JedisConnectionException e) {
e.printStackTrace();
}

return false;
}

public String getValue(String key) {
return pool.getResource().get(key);
}

public void subscribe(String... channels) {
while (!exit) {
try {
pool.getResource().subscribe(pubsub, channels);
} catch (JedisConnectionException e) {
e.printStackTrace();
System.out.println("try reconnect");
try {
Thread.sleep(5000);
} catch (InterruptedException e1) {
e1.printStackTrace();
}
}
}
}
}

Test测试函数:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
package cn.cicoding.utils;

import redis.clients.jedis.JedisPubSub;

public class Test extends JedisPubSub {
@Override
public void onMessage(String channel, String message) {
System.out.println(channel + ":" + message);
}

public static void main(String[] args) {
MQClient client = new MQClient("127.0.0.1", 6379, new Test());
client.setValue("abc", "java setted");
System.out.println(client.getValue("abc"));
System.out.println(client.clientNotify("nodeSubscriber", "message from java", "{\"debug\":0}", MQClient.MSG_REALTIME));
client.subscribe("testInitiativePerception");
}
}

相关文章