import org.fusesource.mqtt.client.Listener; //导入依赖的package包/类
private Listener connectionListener() {
	return new Listener() {

		@Override
		public void onPublish(UTF8Buffer topic, Buffer body, Runnable ack) {
			// nothing to do
		}

		@Override
		public void onFailure(Throwable value) {
			// nothing to do
		}

		@Override
		public void onDisconnected() {
			fireConnectionLost();
		}

		@Override
		public void onConnected() {
			fireReconnected();
		}
	};
}
 

import org.fusesource.mqtt.client.Listener; //导入依赖的package包/类
private void setListener(final boolean sampleByTime, final int sampleCount) {
	connection.listener(new Listener() {
		@Override
		public void onPublish(UTF8Buffer topic, Buffer body, Runnable ack) {
			try {
				ByteArrayOutputStream baos = new ByteArrayOutputStream();
				body.writeTo(baos);
				String msg = baos.toString();
				ack.run();
				
				if(sampleByTime) {
					synchronized (dataLock) {
						handleSubBean(sampleByTime, msg, sampleCount);
					}
				} else {
					synchronized (dataLock) {
						SubBean bean = handleSubBean(sampleByTime, msg, sampleCount);
						//logger.info(System.currentTimeMillis() + ": need notify? receivedCount=" + bean.getReceivedCount() + ", sampleCount=" + sampleCount);
						if(bean.getReceivedCount() == sampleCount) {
							dataLock.notify();
						}
					}
				}
			} catch (IOException e) {
				logger.log(Priority.ERROR, e.getMessage(), e);
			}
		}

		@Override
		public void onFailure(Throwable value) {
			connectFailed = true;
			connection.kill(null);
		}

		@Override
		public void onDisconnected() {
		}

		@Override
		public void onConnected() {
		}
	});
}