Tair (Redis OSS-compatible) mendukung fitur Redis Pub/Sub. Satu klien mempublikasikan pesan ke sebuah channel, sedangkan klien lain dapat berlangganan untuk menerimanya.
Cara kerja Pub/Sub
Pesan yang dipublikasikan oleh Tair (Redis OSS-compatible) bersifat non-persisten dan mengikuti model fire-and-forget. Publisher tidak memverifikasi keberadaan subscriber maupun menyimpan pesan. Subscriber hanya menerima pesan yang dipublikasikan setelah mereka berlangganan.
Publisher tidak memerlukan koneksi eksklusif—Anda dapat menggunakan koneksi yang sama untuk operasi lain saat mempublikasikan. Namun, subscriber memerlukan koneksi eksklusif karena proses tersebut akan diblokir selama menunggu pesan. Gunakan koneksi atau thread terpisah untuk setiap subscriber.
Contoh kode
Publisher (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);
// Password instans
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");
// Publisher menghentikan pengiriman pesan dengan mengirimkan pesan "quit".
jedis.publish(channel, "quit");
}
}Subscriber (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);
// Password instans
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();
// Saat mendengarkan pesan yang telah dilanggani, receiver memblokir proses hingga menerima pesan "quit" (pasif) atau secara aktif berhenti berlangganan.
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();
}
}
}Message listener
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();
// Ketika pesan yang diterima adalah "quit", berhenti berlangganan dari channel tersebut (pasif).
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
}
}Program utama untuk demo
package message.kvstore.aliyun.com;
import java.util.UUID;
import redis.clients.jedis.JedisPubSub;
public class KVStorePubSubTest {
// Informasi koneksi instans, yang dapat diperoleh dari Konsol.
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-Channel-A";
// Publisher mulai mengirim pesan. Pada titik ini, belum ada yang berlangganan, sehingga pesan ini tidak akan diterima.
pubClient.pub(channel, "Alibaba Cloud Message 1: (No one has subscribed yet, so this message will not be received)");
// Subscriber pesan.
KVStoreSubClient subClient = new KVStoreSubClient(host, port,password);
JedisPubSub listener = new KVStoreMessageListener();
subClient.setChannelAndListener(listener, channel);
// Subscriber mulai berlangganan.
subClient.start();
// Publisher melanjutkan pengiriman pesan.
for (int i = 0; i < 5; i++) {
String message=UUID.randomUUID().toString();
pubClient.pub(channel, message);
Thread.sleep(1000);
}
// Subscriber secara aktif berhenti berlangganan.
subClient.unsubscribe(channel);
Thread.sleep(1000);
pubClient.pub(channel, "Alibaba Cloud Message 2: (The subscription is canceled, so this message will not be received)");
// Publisher menghentikan pengiriman pesan dengan mengirimkan pesan "quit".
// Jika terdapat subscriber lain, mereka akan menjalankan operasi "unsubscribe" ketika menerima "quit" di listener.onMessage().
pubClient.close(channel);
}
}Contoh output
Jalankan program Java dengan endpoint dan password yang benar untuk instans Tair (Redis OSS-compatible) Anda. Output yang diharapkan:
>>> 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:quitContoh ini menggunakan satu publisher dan satu subscriber. Modifikasi kode sesuai kebutuhan untuk mendukung multiple publisher, subscriber, dan channel.