RefreshActive refreshZkActive = new RefreshActive(active, conf,
zkCluster, topologyId, zkActive);
refreshZkActive.run();
// ���������߳�
RunnableCallback heartbeat_fn = new WorkerHeartbeatRunable(conf,
workerId, port, topologyId, new CopyOnWriteArraySet<Integer>(taskids),
active);
heartbeat_fn.run();
// ����worker����tuple�Ļ�����
LinkedBlockingQueue<TransferData> transferQueue = new LinkedBlockingQueue<TransferData>();