## Xây dựng hệ thống nhắn tin thời gian thực

Hệ thống Pub/Sub của Redis có thể xây dựng các hệ thống nhắn tin thời gian thực, ví dụ như nhiều nhà phát triển sử dụng Pub/Sub để xây dựng hệ thống chat thời gian thực.

``` java
import redis.clients.jedis.*;
import java.util.Date;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.commons.lang3.RandomStringUtils;
class PrintListener extends JedisPubSub{
    @Override
    public void onMessage(String channel, String message) {
        String time = DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss");
        System.out.println("message receive:" + message + ",channel:" + channel +
        "..." + time);
        //Tại đây chúng ta có thể hủy đăng ký (unsubscribe)
        if(message.equalsIgnoreCase("quit")){
            this.unsubscribe(channel);
        }
    }
}
class PubClient {
    private Jedis jedis;
    public PubClient(String host,int port){
        jedis = new Jedis(host,port);
    }
    public void pub(String channel,String message){
        jedis.publish(channel, message);
    }
    public void close(String channel){
        jedis.publish(channel, "quit");
        jedis.del(channel);//Hệ thống nhắn tin thời gian thực
    }
}
class SubClient {
    private Jedis jedis;//
    public SubClient(String host,int port){
        jedis = new Jedis(host,port);
    }
    public void sub(JedisPubSub listener,String channel){
        jedis.subscribe(listener, channel);
        //Nơi này sẽ bị chặn (blocked), ở cấp độ mã client cho JedisPubSub trong việc xử lý tin nhắn, sẽ "độc chiếm" liên kết
        //và phương thức vòng lặp while được sử dụng, lắng nghe các tin nhắn đã đăng ký
    }
}
public class PubSubTest {
    /**
    * @param args
    */
    static String host = "127.0.0.1";
    static int port = 10011;
    public static void main(String[] args) throws Exception{
    PubClient pubClient = new PubClient(host,port);
    final String channel = "pubsub-channel";
    pubClient.pub(channel, "before1");
    pubClient.pub(channel, "before2");
    Thread.sleep(2000);
    //Trình đăng ký nhận tin nhắn rất đặc biệt và cần độc chiếm kết nối, vì vậy chúng ta cần tạo một kết nối mới cho nó;
    //ngoài ra, việc triển khai phía client cũng đảm bảo tính năng "độc chiếm liên kết", phương thức sub sẽ tiếp tục chặn,
    //cho đến khi phương thức listener.unsubscribe được gọi 
    Thread subThread = new Thread(new Runnable() {
        @Override
        public void run() {
            try{
                SubClient subClient = new SubClient(host,port);
                System.out.println("----------subscribe operation begin-------");
                JedisPubSub listener = new PrintListener();
                //Ở cấp độ API, đây là một hoạt động thăm dò (polling), và nó sẽ chỉ trả về khi unsubscribe được gọi
                subClient.sub(listener, channel);
                System.out.println("----------subscribe operation end-------")
                ;
            }catch(Exception e){
                e.printStackTrace();
            }
        }
    });
    subThread.start();
    int i=0;
    while(i < 10){
        String message = RandomStringUtils.random(6, true, true);//apache-commons
        pubClient.pub(channel, message);
        i++;
        Thread.sleep(1000);
    }
    //Chỉ thị đóng thụ động, nếu phía xuất bản (publisher) trong kênh xác nhận rằng kênh cần được đóng, thì gửi một "quit"
    //sau đó khi listener.onMessage() nhận được "quit", các client đăng ký khác sẽ thực hiện thao tác "unsubscribe".
    pubClient.close(channel);
    //Ngoài ra, bạn có thể hủy đăng ký như thế này
    //listener.unsubscribe(channel);
    }
}
```

Kết quả đầu ra:

```
----------subscribe operation begin-------
message receive:erRIEe,channel:pubsub-channel...2016-03-15 15:53:52
message receive:Ovcwiw,channel:pubsub-channel...2016-03-15 15:53:53
message receive:STPWfV,channel:pubsub-channel...2016-03-15 15:53:54
message receive:SR4iIk,channel:pubsub-channel...2016-03-15 15:53:55
message receive:GI3Ege,channel:pubsub-channel...2016-03-15 15:53:56
message receive:0V1JUt,channel:pubsub-channel...2016-03-15 15:53:57
message receive:3iU8BV,channel:pubsub-channel...2016-03-15 15:53:58
message receive:BqeI2x,channel:pubsub-channel...2016-03-15 15:53:59
message receive:D53cHF,channel:pubsub-channel...2016-03-15 15:54:00
message receive:quit,channel:pubsub-channel...2016-03-15 15:54:01
----------subscribe operation end-------
```