mysql消息订阅与发布_消息发布与订阅代码⽰例
消息发布者 (即publish client)
package message.;
import redis.clients.jedis.Jedis;
public class KVStorePubClient {
private Jedis jedis;
public KVStorePubClient(String host,int port, String password){
jedis = new Jedis(host,port);
//KVStore的实例密码
String authString = jedis.auth(password);
if (!authString.equals("OK"))
{
return;
}
}
public void pub(String channel,String message){
System.out.println(" >>> 发布(PUBLISH) > Channel:"+channel+" > 发送出的Message:"+message);联通流量领取
jedis.publish(channel, message);
}
public void clo(String channel){
System.out.println(" >>> 发布(PUBLISH)结束 > Channel:"+channel+" > Message:quit");
/
/消息发布者结束发送，即发送⼀个“quit”消息；
jedis.publish(channel, "quit");
}
}
消息订阅者 (即subscribe client)
package message.;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPubSub;
public class KVStoreSubClient extends Thread{
private Jedis jedis;
private String channel;
private JedisPubSub listener;
public KVStoreSubClient(String host,int port, String password){
jedis = new Jedis(host,port);
//ApsaraDB for Redis的实例密码
String authString = jedis.auth(password);//password
if (!authString.equals("OK"))
{
return;
}
}
public void tChannelAndListener(JedisPubSub listener,String channel){
this.listener=listener;
this.channel=channel;
}
private void subscribe(){
if(listener==null || channel==null){
}
System.out.println(" >>> 订阅(SUBSCRIBE) > Channel:"+channel);
System.out.println();
//接收者在侦听订阅的消息时，将会阻塞进程，直⾄接收到quit消息(被动⽅式)，或主动取消订阅jedis.subscribe(listener, channel);
}
public void unsubscribe(String channel){
System.out.println(" >>> 取消订阅(UNSUBSCRIBE) > Channel:"+channel);
System.out.println();
listener.unsubscribe(channel);
}
@Override
public void run() {
try{
System.out.println();
System.out.println("----------订阅消息SUBSCRIBE 开始-------");
subscribe();
System.out.println("----------订阅消息SUBSCRIBE 结束-------");
System.out.println();东北酸菜饺子
}catch(Exception e){
e.printStackTrace();
}
}
}
消息监听者
package message.;
排球的起源
import redis.clients.jedis.JedisPubSub;
public class KVStoreMessageListener extends JedisPubSub{
@Override
public void onMessage(String channel, String message) {
System.out.println(" <<< 订阅(SUBSCRIBE)< Channel:" + channel + " >接收到的Message:" + message ); System.out.println();
曹操几个儿子//当接收到的message为quit时，取消订阅(被动⽅式)
if(message.equalsIgnoreCa("quit")){
this.unsubscribe(channel);
}
}
@Override
public void onPMessage(String pattern, String channel, String message) {
// TODO Auto-generated method stub
}
@Override
public void onSubscribe(String channel, int subscribedChannels) {
// TODO Auto-generated method stub
}
@Override
public void onUnsubscribe(String channel, int subscribedChannels) {
// TODO Auto-generated method stub
}
@Override
public void onPUnsubscribe(String pattern, int subscribedChannels) {
// TODO Auto-generated method stub
}
@Override
public void onPSubscribe(String pattern, int subscribedChannels) {
// TODO Auto-generated method stub
}
}
⽰例主程序
package message.;
import java.util.UUID;
import redis.clients.jedis.JedisPubSub;
public class KVStorePubSubTest {
//ApsaraDB for Redis的连接信息，从控制台可以获得
static final String host = "xxxxxxxxxx.mhza.";
static final int port = 6379;
static final String password="password";//password
public static void main(String[] args) throws Exception{
KVStorePubClient pubClient = new KVStorePubClient(host, port,password); final String channel = "KVStore频道-A";
//消息发送者开始发消息，此时还⽆⼈订阅，所以此消息不会被接收
pubClient.pub(channel, "Aliyun消息1：(此时还⽆⼈订阅，所以此消息不会被接收)"); //消息接收者
KVStoreSubClient subClient = new KVStoreSubClient(host, port,password); JedisPubSub listener = new KVStoreMessageListener();
网速慢怎么解决subClient.tChannelAndListener(listener, channel);
//消息接收者开始订阅
subClient.start();
//消息发送者继续发消息
for (int i = 0; i < 5; i++) {
String message=UUID.randomUUID().toString();
夕阳的意思pubClient.pub(channel, message);
Thread.sleep(1000);
}
//消息接收者主动取消订阅
步辇图作者subClient.unsubscribe(channel);
Thread.sleep(1000);
pubClient.pub(channel, "Aliyun消息2：(此时订阅取消，所以此消息不会被接收)");
//消息发布者结束发送，即发送⼀个“quit”消息；
123英文//此时如果有其他的消息接收者，那么在Message()中接收到“quit”时，将执⾏“unsubscribe”操作。pubClient.clo(channel);
}
}
