Tair (Redis OSS-compatible) prend en charge la fonctionnalité Redis Pub/Sub. Un client publie des messages sur un canal, tandis que d'autres clients s'y abonnent pour les recevoir.
Fonctionnement de Pub/Sub
Les messages publiés par Tair (Redis OSS-compatible) ne sont pas persistants et suivent le modèle « fire-and-forget » (envoyer et oublier). L'éditeur ne vérifie pas la présence d'abonnés et n'enregistre pas les messages. Les abonnés ne reçoivent que les messages publiés après leur abonnement.
Un éditeur n'a pas besoin d'une connexion exclusive : vous pouvez utiliser la même connexion pour d'autres opérations pendant la publication. En revanche, un abonné nécessite une connexion exclusive, car il reste bloqué en attendant les messages. Utilisez une connexion ou un thread distinct pour chaque abonné.
Exemples de code
Éditeur (client de publication)
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");
}
}Abonné (client d'abonnement)
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();
}
}
}Écouteur de messages
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
}
}Programme principal de la démonstration
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);
}
}Sortie exemple
Exécutez le programme Java avec l'endpoint et le mot de passe corrects pour votre instance Tair (Redis OSS-compatible). Voici la sortie attendue :
>>> 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:quitCet exemple utilise un seul éditeur et un seul abonné. Modifiez le code selon vos besoins pour prendre en charge plusieurs éditeurs, abonnés et canaux.