Tair (Redis OSS-compatible) menyediakan fitur publikasi dan langganan pesan yang serupa dengan Redis sumber terbuka. Fitur ini memungkinkan beberapa klien untuk berlangganan pesan yang diterbitkan oleh satu klien.
Informasi latar belakang
Pesan yang diterbitkan oleh Tair (Redis OSS-compatible) bersifat tidak persisten. Penerbit hanya bertanggung jawab untuk menerbitkan pesan, tanpa memperhatikan apakah pesan tersebut diterima atau disimpan. Dengan demikian, pesan akan hilang setelah diterbitkan. Pelanggan hanya dapat menerima pesan yang dikirim setelah mereka berlangganan ke penerbit. Mereka tidak akan menerima pesan sebelumnya di saluran.
Selain itu, klien penerbit tidak harus terhubung secara eksklusif ke server. Saat menerbitkan pesan, klien dapat melakukan operasi lain seperti operasi Daftar dari klien yang sama. Namun, klien pelanggan perlu terhubung secara eksklusif ke server. Selama periode langganan, klien tidak dapat melakukan operasi lain karena operasi diblokir saat menunggu pesan di saluran. Oleh karena itu, klien pelanggan harus menggunakan server khusus atau thread terpisah untuk menerima pesan (lihat contoh berikut).
Contoh kode
Untuk penerbit pesan (klien penerbit):
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);
// Kata sandi
String authString = jedis.auth(password);
if (!authString.equals("OK"))
{
System.err.println("AUTH Gagal: " + authString);
return;
}
}
public void pub(String channel,String message){
System.out.println(" >>> PUBLISH > Saluran:"+channel+" > Pesan terkirim: "+message);
jedis.publish(channel, message);
}
public void close(String channel){
System.out.println(" >>> PUBLISH berakhir > Saluran: "+channel+" > Pesan:quit");
//Penerbit pesan mengakhiri proses dengan mengirim pesan keluar.
jedis.publish(channel, "quit");
}
}Untuk pelanggan pesan (klien pelanggan):
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);
// Kata sandi
String authString = jedis.auth(password);//password
if (!authString.equals("OK"))
{
System.err.println("AUTH Gagal: " + 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("Kesalahan:SubClient> pendengar atau saluran bernilai null");
}
System.out.println(" >>> BERLANGGANAN > Saluran:"+channel);
System.out.println();
//Saat penerima mendengarkan pesan yang dilanggan, proses diblokir sampai pesan keluar diterima (secara pasif) atau langganan dibatalkan secara aktif.
jedis.subscribe(listener, channel);
}
public void unsubscribe(String channel){
System.out.println(" >>> BATAL BERLANGGANAN > Saluran:"+channel);
System.out.println();
listener.unsubscribe(channel);
}
@Override
public void run() {
try{
System.out.println();
System.out.println("---------BERLANGGANAN dimulai-------");
subscribe();
System.out.println("----------BERLANGGANAN berakhir-------");
System.out.println();
}catch(Exception e){
e.printStackTrace();
}
}
}Untuk pendengar pesan:
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(" <<< BERLANGGANAN < Saluran: " + channel + "> Pesan diterima: " + message );
System.out.println();
//Saat pesan keluar diterima, langganan dibatalkan (secara 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
}
}Blok kode utama:
package message.kvstore.aliyun.com;
import java.util.UUID;
import redis.clients.jedis.JedisPubSub;
public class KVStorePubSubTest {
//Informasi koneksi instance. Informasi ini 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";
//Pengirim pesan mulai mengirim pesan, tetapi belum ada klien yang berlangganan ke saluran. Oleh karena itu, pesan tidak akan diterima.
pubClient.pub(channel, "Pesan Alibaba Cloud 1: (Tidak ada pelanggan. Pesan ini tidak akan diterima.)");
//Penerima pesan.
KVStoreSubClient subClient = new KVStoreSubClient(host, port,password);
JedisPubSub listener = new KVStoreMessageListener();
subClient.setChannelAndListener(listener, channel);
//Penerima pesan berlangganan pesan.
subClient.start();
//Pengirim pesan melanjutkan pengiriman pesan.
for (int i = 0; i < 5; i++) {
String message=UUID.randomUUID().toString();
pubClient.pub(channel, message);
Thread.sleep(1000);
}
//Penerima pesan membatalkan langganan pesan.
subClient.unsubscribe(channel);
Thread.sleep(1000);
pubClient.pub(channel, "Pesan Alibaba Cloud 2:(Langganan dibatalkan. Pesan ini tidak akan diterima.)");
//Penerbit pesan mengakhiri proses dengan mengirim pesan keluar.
//Saat penerima pesan lainnya menerima pesan keluar di listener.onMessage(), operasi UNSUBSCRIBE dilakukan.
pubClient.close(channel);
}
}Hasil
Setelah memasukkan titik akhir dan kata sandi yang benar untuk terhubung ke instance Tair (Redis OSS-compatible) dan menjalankan kode Java di atas, hasil berikut ditampilkan:
>>> PUBLISH > Saluran:KVStore Channel-A > Pesan terkirim: Pesan Alibaba Cloud 1: (Tidak ada pelanggan. Pesan ini tidak akan diterima.)
----------BERLANGGANAN dimulai-------
>>> BERLANGGANAN > Saluran: KVStore Channel-A
>>> PUBLISH > Saluran: KVStore Channel-A > Pesan terkirim: 0f9c2cee-77c7-4498-89a0-1dc5a2f65889
<<< BERLANGGANAN < Saluran:KVStore Channel-A > Pesan diterima: 0f9c2cee-77c7-4498-89a0-1dc5a2f65889
>>> PUBLISH > Saluran: KVStore Channel-A > Pesan terkirim: ed5924a9-016b-469b-8203-7db63d06f812
<<< BERLANGGANAN < Saluran:KVStore Channel-A > Pesan diterima: ed5924a9-016b-469b-8203-7db63d06f812
>>> PUBLISH > Saluran: KVStore Channel-A > Pesan terkirim: f1f84e0f-8f35-4362-9567-25716b1531cd
<<< BERLANGGANAN < Saluran:KVStore Channel-A > Pesan diterima: f1f84e0f-8f35-4362-9567-25716b1531cd
>>> PUBLISH > Saluran: KVStore Channel-A > Pesan terkirim: 746bde54-af8f-44d7-8a49-37d1a245d21b
<<< BERLANGGANAN < Saluran:KVStore Channel-A > Pesan diterima: 746bde54-af8f-44d7-8a49-37d1a245d21b
>>> PUBLISH > Saluran: KVStore Channel-A > Pesan terkirim: 8ac3b2b8-9906-4f61-8cad-84fc1f15a3ef
<<< BERLANGGANAN < Saluran:KVStore Channel-A > Pesan diterima: 8ac3b2b8-9906-4f61-8cad-84fc1f15a3ef
>>> BATAL BERLANGGANAN > Saluran: KVStore Channel-A
----------BERLANGGANAN berakhir-------
>>> PUBLISH > Saluran:KVStore Channel-A > Pesan terkirim: Pesan Alibaba Cloud 2: (Langganan dibatalkan. Pesan ini tidak akan diterima.)
>>> PUBLISH berakhir > Saluran:KVStore Channel-A > Pesan:quitContoh di atas menunjukkan skenario dengan satu penerbit dan satu pelanggan. Namun, dapat ada beberapa penerbit, pelanggan, bahkan beberapa saluran pesan. Anda dapat menyesuaikan kode untuk mendukung skenario tersebut.