很簡單，看例子

import java.util.List;
import com.aliyun.drc.client.message.DataMessage;
import com.aliyun.drc.clusterclient.ClusterClient;
import com.aliyun.drc.clusterclient.ClusterListener;
import com.aliyun.drc.clusterclient.DefaultClusterClient;
import com.aliyun.drc.clusterclient.RegionContext;
import
 
 com.aliyun.drc.clusterclient.message.ClusterMessage;

/**
 * 其中AccessKey、Secret可在阿里雲使用者介面中免費獲取 GUID需要開通訂閱服務才能獲取對應的資料變更監聽服務
 * 
 */
public class TestDTS {
    public static void main(String[] args) throws Exception {
        RegionContext context = new RegionContext();
        context.setUsePublicIp(true
 
);
        context.setAccessKey("***key");
        context.setSecret("***secret");
        ClusterClient client = new DefaultClusterClient(context);
        ClusterListener listener = new ClusterListener() {
            public void notify(List<ClusterMessage> list) throws Exception {
                
 
for (ClusterMessage m : list) {
                    if (m.getRecord().getOpt().compareTo(DataMessage.Record.Type.INSERT) == 0
                            || m.getRecord().getOpt().compareTo(DataMessage.Record.Type.REPLACE) == 0
                            || m.getRecord().getOpt().compareTo(DataMessage.Record.Type.UPDATE) == 0) {
                        try {
                            System.out.println("xxx");
                        } catch (Exception e) {
                        }
                    }
                }
            }
            public void noException(Exception e) {
            }
        };

        client.addConcurrentListener(listener);
        client.askForGUID("***guid");
        client.start();
    }
}

可以用一個阻塞佇列做生產者消費者模型，加上快取等策略，達到接收訂閱變化，準實時反映到後續搜尋引擎、訊息佇列或其他需求方中

public static LinkedBlockingQueue<String> operationQueue = new LinkedBlockingQueue<String>();

    static class Worker implements Runnable {
        @Override
        public void run()  {
            try {
                while(true) {
                   String oper = operationQueue.take();
                   // do something comsume
                }
            } catch (Exception ex) {
            }
        }
    }