阿里云数据变化订阅

很简单，看例子

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) {
            }
        }
    }