代码示例

消息发布者 (即publish client)

package message.kvstore.aliyun.com;

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"))

{

System.err.println("AUTH Failed: " + authString);

return;

}

}

public void pub(String channel,String message){

System.out.println(" >>> 发布(PUBLISH) > Channel:"+channel+" > 发送出的Message:"+message);

jedis.publish(channel, message);

}

public void close(String channel){

System.out.println(" >>> 发布(PUBLISH)结束 > Channel:"+channel+" > Message:quit");

//消息发布者结束发送，即发送一个“quit”消息；

jedis.publish(channel, "quit");

}

}

消息订阅者 (即subscribe client)

package message.kvstore.aliyun.com;

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"))

{

System.err.println("AUTH Failed: " + authString);

return;

}

}

public void setChannelAndListener(JedisPubSub listener,String channel){

this.listener=listener;

this.channel=channel;

}

private void subscribe(){

if(listener==null || channel==null){

System.err.println("Error:SubClient> listener or channel is 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.kvstore.aliyun.com;

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.equalsIgnoreCase("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.kvstore.aliyun.com;

import java.util.UUID;

import redis.clients.jedis.JedisPubSub;

public class KVStorePubSubTest {

//ApsaraDB for Redis的连接信息，从控制台可以获得

static final String host = "xxxxxxxxxx.m.cnhza.kvstore.aliyuncs.com";

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.setChannelAndListener(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”消息；

//此时如果有其他的消息接收者，那么在listener.onMessage()中接收到“quit”时，将执行“unsubscribe”操作。

pubClient.close(channel);

}

}