import com.aliyun.openservices.ons.api.Consumer; //导入依赖的package包/类
private void initializeConsumers(ConsumerMeta meta) throws InterruptedException, MQClientException {
		Assert.hasText(meta.getConsumerId(), "consumerId can not be empty!");
		Assert.hasText(meta.getTopic(), "topic can not be empty!");
		logger.info("create mq consumerId: {}", meta.getConsumerId());

		Properties comsumerProperties = onsProperties.baseProperties();
		comsumerProperties.setProperty(PropertyKeyConst.ConsumerId, meta.getConsumerId());
		comsumerProperties.setProperty(PropertyKeyConst.MessageModel, meta.getMessageModel().name());
		comsumerProperties.setProperty(PropertyKeyConst.MaxReconsumeTimes, String.valueOf(meta.getMaxReconsumeTimes()));
		Properties customProps = onsProperties.getConsumers().get(meta.getConsumerId());
		if(customProps!=null){
			comsumerProperties.putAll(customProps);
		}
//		rawConsumer.setMessageModel(meta.getMessageModel());
		
		Consumer consumer = ONSFactory.createConsumer(comsumerProperties);
		DefaultMQPushConsumer rawConsumer = (DefaultMQPushConsumer)ReflectUtils.getFieldValue(consumer, "defaultMQPushConsumer");
		rawConsumer.setConsumeFromWhere(meta.getConsumeFromWhere());
		
//		consumer.subscribe(meta.getTopic(), meta.getSubExpression(), listener);
		ListenerType listenerType = meta.getListenerType(); 
		if(listenerType==ListenerType.CUSTOM){
			rawConsumer.subscribe(meta.getTopic(), meta.getSubExpression());
			registerONSConsumerListener(rawConsumer, meta);
			rawConsumer.start();
		}else if(listenerType==ListenerType.RMQ){
			rawConsumer.subscribe(meta.getTopic(), meta.getSubExpression());
			rawConsumer.registerMessageListener((MessageListenerConcurrently)meta.getListener());
			rawConsumer.start();
		}else{
			consumer.subscribe(meta.getTopic(), meta.getSubExpression(), (MessageListener)meta.getListener());
			consumer.start();
		}
		logger.info("ONSConsumer[{}] started! meta: {}", meta.getConsumerId(), meta);
	}
 

import com.aliyun.openservices.ons.api.Consumer; //导入依赖的package包/类
public static synchronized Consumer mkInstance(ConsumerConfig consumerConfig, MessageListener listener) throws Exception {
	

	String consumerId = consumerConfig.getConsumerId();
	Consumer consumer = consumers.get(consumerId);
	if (consumer != null) {

		LOG.info("Consumer of " + consumerId + " has been created, don't recreate it ");

		// Attention, this place return null to info duplicated consumer
		return null;
	}

	Properties properties = new Properties();
	properties.put(PropertyKeyConst.AccessKey, consumerConfig.getAccessKey());
	properties.put(PropertyKeyConst.SecretKey, consumerConfig.getSecretKey());
	properties.put(PropertyKeyConst.ConsumerId, consumerId);
	properties.put(PropertyKeyConst.ConsumeThreadNums, consumerConfig.getConsumerThreadNum());
	consumer = ONSFactory.createConsumer(properties);

	consumer.subscribe(consumerConfig.getTopic(), consumerConfig.getSubExpress(), listener);
	consumer.start();

	consumers.put(consumerId, consumer);
	LOG.info("Successfully create " + consumerId + " consumer");

	return consumer;

}
 

import com.aliyun.openservices.ons.api.Consumer; //导入依赖的package包/类
public static synchronized Consumer mkInstance(ConsumerConfig consumerConfig, MessageListener listener) throws Exception {
    String consumerId = consumerConfig.getConsumerId();
    Consumer consumer = consumers.get(consumerId);
    if (consumer != null) {

        LOG.info("Consumer of " + consumerId + " has been created, don't recreate it ");

        // Attention, this place return null to info duplicated consumer
        return null;
    }

    Properties properties = new Properties();
    properties.put(PropertyKeyConst.AccessKey, consumerConfig.getAccessKey());
    properties.put(PropertyKeyConst.SecretKey, consumerConfig.getSecretKey());
    properties.put(PropertyKeyConst.ConsumerId, consumerId);
    properties.put(PropertyKeyConst.ConsumeThreadNums, consumerConfig.getConsumerThreadNum());
    consumer = ONSFactory.createConsumer(properties);

    consumer.subscribe(consumerConfig.getTopic(), consumerConfig.getSubExpress(), listener);
    consumer.start();

    consumers.put(consumerId, consumer);
    LOG.info("Successfully create " + consumerId + " consumer");

    return consumer;

}
 

import com.aliyun.openservices.ons.api.Consumer; //导入依赖的package包/类
@Override
public void run() {
    ConsumerOptional consumerOptional = consumerId.getConsumerOptional();

    String cid = consumerId.getCid() + suffix;

    String topic = consumerId.getTopic();
    if (null == topic) {
        throw new RuntimeException(String.format("%s 必须订阅一个topic", consumerId.getCid()));
    }
    topic += suffix;

    Properties properties = new Properties();
    properties.put(PropertyKeyConst.ConsumerId, cid);
    properties.put(PropertyKeyConst.AccessKey, accessKey);
    properties.put(PropertyKeyConst.SecretKey, secretKey);
    properties.put(PropertyKeyConst.ConsumeThreadNums, consumerOptional.getConsumeThread());
    properties.put(PropertyKeyConst.MessageModel, consumerOptional.getConsumerModel());
    properties.put(PropertyKeyConst.MaxReconsumeTimes, consumerOptional.getMaxReconsume());
    properties.put(PropertyKeyConst.SuspendTimeMillis,consumerOptional.getSuspendTime());
    String tags = consumerId.getTags();

    if(tags.contains("||")){
        tags = tags.substring(0,tags.length()-2);
    }
    logger.info("发现频道 CID: {}  订阅TOPIC: {}  TAG: {}", cid, topic, tags);

    if (consumerId.isOrdered()) {
        properties.put(PropertyKeyConst.SuspendTimeMillis, consumerOptional.getSuspendTime());
        OrderConsumer orderConsumer = ONSFactory.createOrderedConsumer(properties);

        orderConsumer.subscribe(topic,tags,((message, context) ->
                dispatch(consumerId, message, consumerOptional.getMaxReconsume()).equals(Action.commit)?
                        OrderAction.Success : OrderAction.Suspend
        ));
        orderConsumer.start();
    } else {
        Consumer consumer = ONSFactory.createConsumer(properties);
        consumer.subscribe(topic, tags, (message, context) ->
                dispatch(consumerId, message, consumerOptional.getMaxReconsume()).equals(Action.commit)?
                        com.aliyun.openservices.ons.api.Action.CommitMessage : com.aliyun.openservices.ons.api.Action.ReconsumeLater
        );
        consumer.start();
   }
    logger.info("消费者启动成功: {}({}) , 消费模式: {} , 消费线程数: {} , 最大重试次数: {} {} , 订阅 TOPIC: {} TAG: {}",
            cid, consumerId.isOrdered()?"有序":"无序",
            consumerOptional.getConsumerModel(), consumerOptional.getConsumeThread(), consumerOptional.getMaxReconsume(),
            consumerId.isOrdered()?", 重试前的等待时间:"+consumerOptional.getSuspendTime()+"毫秒":"",topic,tags);
}