Tair (Redis OSS-compatible) admite la función Pub/Sub de Redis. Un cliente publica mensajes en un canal y otros clientes se suscriben para recibirlos.
Funcionamiento de Pub/Sub
Los mensajes que publica Tair (Redis OSS-compatible) no son persistentes y siguen un modelo de envío sin confirmación. El publicador no verifica si hay suscriptores ni guarda los mensajes. Los suscriptores solo reciben los mensajes publicados después de su suscripción.
El publicador no requiere una conexión exclusiva; puede usar la misma conexión para otras operaciones mientras publica. Sin embargo, el suscriptor necesita una conexión exclusiva porque se bloquea a la espera de mensajes. Use una conexión o hilo independiente para cada suscriptor.
Ejemplos de código
Publicador (cliente de publicación)
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);
// The password of the instance
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+" > Sent Message:"+message);
jedis.publish(channel, message);
}
public void close(String channel){
System.out.println(" >>> PUBLISH ends > Channel:"+channel+" > Message:quit");
// The publisher stops sending messages by sending a "quit" message.
jedis.publish(channel, "quit");
}
}Suscriptor (cliente de suscripción)
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);
// The password of the instance
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();
// When listening for subscribed messages, the receiver blocks the process until it receives a "quit" message (passive) or actively unsubscribes.
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 starts-------");
subscribe();
System.out.println("----------SUBSCRIBE ends-------");
System.out.println();
}catch(Exception e){
e.printStackTrace();
}
}
}Escucha de mensajes
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 + " > Received Message:" + message );
System.out.println();
// When the received message is "quit", unsubscribe from the channel (passive).
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
}
}Programa principal de la demostración
package message.kvstore.aliyun.com;
import java.util.UUID;
import redis.clients.jedis.JedisPubSub;
public class KVStorePubSubTest {
// The connection information of the instance, which can be obtained from the console.
static final String host = "xxxxxxxxxx.m.cnhza.kvstore.aliyuncs.com";
static final int port = 6379;
static final String password="password";// The password.
public static void main(String[] args) throws Exception{
KVStorePubClient pubClient = new KVStorePubClient(host, port,password);
final String channel = "KVStore-Channel-A";
// The publisher starts sending messages. At this point, no one has subscribed, so this message will not be received.
pubClient.pub(channel, "Alibaba Cloud Message 1: (No one has subscribed yet, so this message will not be received)");
// The message subscriber.
KVStoreSubClient subClient = new KVStoreSubClient(host, port,password);
JedisPubSub listener = new KVStoreMessageListener();
subClient.setChannelAndListener(listener, channel);
// The subscriber starts to subscribe.
subClient.start();
// The publisher continues to send messages.
for (int i = 0; i < 5; i++) {
String message=UUID.randomUUID().toString();
pubClient.pub(channel, message);
Thread.sleep(1000);
}
// The subscriber actively unsubscribes.
subClient.unsubscribe(channel);
Thread.sleep(1000);
pubClient.pub(channel, "Alibaba Cloud Message 2: (The subscription is canceled, so this message will not be received)");
// The publisher stops sending messages by sending a "quit" message.
// If there are other subscribers, they will execute the "unsubscribe" operation when they receive "quit" in listener.onMessage().
pubClient.close(channel);
}
}Salida de ejemplo
Ejecute el programa Java con el endpoint y la contraseña correctos para su instancia de Tair (Redis OSS-compatible). Salida esperada:
>>> PUBLISH > Channel:KVStore-Channel-A > Sent Message:Alibaba Cloud Message 1: (No one has subscribed yet, so this message will not be received)
----------SUBSCRIBE starts-------
>>> SUBSCRIBE > Channel:KVStore-Channel-A
>>> PUBLISH > Channel:KVStore-Channel-A > Sent Message:0f9c2cee-77c7-4498-89a0-1dc5a2f65889
<<< SUBSCRIBE < Channel:KVStore-Channel-A > Received Message:0f9c2cee-77c7-4498-89a0-1dc5a2f65889
>>> PUBLISH > Channel:KVStore-Channel-A > Sent Message:ed5924a9-016b-469b-8203-7db63d06f812
<<< SUBSCRIBE < Channel:KVStore-Channel-A > Received Message:ed5924a9-016b-469b-8203-7db63d06f812
>>> PUBLISH > Channel:KVStore-Channel-A > Sent Message:f1f84e0f-8f35-4362-9567-25716b1531cd
<<< SUBSCRIBE < Channel:KVStore-Channel-A > Received Message:f1f84e0f-8f35-4362-9567-25716b1531cd
>>> PUBLISH > Channel:KVStore-Channel-A > Sent Message:746bde54-af8f-44d7-8a49-37d1a245d21b
<<< SUBSCRIBE < Channel:KVStore-Channel-A > Received Message:746bde54-af8f-44d7-8a49-37d1a245d21b
>>> PUBLISH > Channel:KVStore-Channel-A > Sent Message:8ac3b2b8-9906-4f61-8cad-84fc1f15a3ef
<<< SUBSCRIBE < Channel:KVStore-Channel-A > Received Message:8ac3b2b8-9906-4f61-8cad-84fc1f15a3ef
>>> UNSUBSCRIBE > Channel:KVStore-Channel-A
----------SUBSCRIBE ends-------
>>> PUBLISH > Channel:KVStore-Channel-A > Sent Message:Alibaba Cloud Message 2: (The subscription is canceled, so this message will not be received)
>>> PUBLISH ends > Channel:KVStore-Channel-A > Message:quitEste ejemplo usa un publicador y un suscriptor. Modifique el código según sea necesario para admitir varios publicadores, suscriptores y canales.