in ons-core/ons-client/src/main/java/org/apache/rocketmq/ons/api/impl/rocketmq/OrderProducerImpl.java [45:97]
public OrderProducerImpl(final Properties properties) {
super(properties);
String producerGroup = properties.getProperty(PropertyKeyConst.GROUP_ID, properties.getProperty(PropertyKeyConst.ProducerId));
if (StringUtils.isEmpty(producerGroup)) {
producerGroup = "__ONS_PRODUCER_DEFAULT_GROUP";
}
this.defaultMQProducer =
new DefaultMQProducer(this.getNamespace(), producerGroup, new OnsClientRPCHook(sessionCredentials));
this.defaultMQProducer.setProducerGroup(producerGroup);
boolean isVipChannelEnabled = Boolean.parseBoolean(properties.getProperty(PropertyKeyConst.isVipChannelEnabled, "false"));
this.defaultMQProducer.setVipChannelEnabled(isVipChannelEnabled);
String sendMsgTimeoutMillis = properties.getProperty(PropertyKeyConst.SendMsgTimeoutMillis, "3000");
this.defaultMQProducer.setSendMsgTimeout(Integer.parseInt(sendMsgTimeoutMillis));
// boolean addExtendUniqInfo = Boolean.parseBoolean(properties.getProperty(PropertyKeyConst.EXACTLYONCE_DELIVERY, "false"));
// this.defaultMQProducer.setAddExtendUniqInfo(addExtendUniqInfo);
if (properties.containsKey(PropertyKeyConst.LANGUAGE_IDENTIFIER)) {
int language = Integer.valueOf(properties.get(PropertyKeyConst.LANGUAGE_IDENTIFIER).toString());
byte languageByte = (byte) language;
this.defaultMQProducer.setLanguage(LanguageCode.valueOf(languageByte));
}
String instanceName = properties.getProperty(PropertyKeyConst.InstanceName, this.buildIntanceName());
this.defaultMQProducer.setInstanceName(instanceName);
this.defaultMQProducer.setNamesrvAddr(this.getNameServerAddr());
String msgTraceSwitch = properties.getProperty(PropertyKeyConst.MsgTraceSwitch);
if (!UtilAll.isBlank(msgTraceSwitch) && (!Boolean.parseBoolean(msgTraceSwitch))) {
LOGGER.info("MQ Client Disable the Trace Hook!");
} else {
try {
Properties tempProperties = new Properties();
tempProperties.put(OnsTraceConstants.AccessKey, sessionCredentials.getAccessKey());
tempProperties.put(OnsTraceConstants.SecretKey, sessionCredentials.getSecretKey());
tempProperties.put(OnsTraceConstants.MaxMsgSize, "128000");
tempProperties.put(OnsTraceConstants.AsyncBufferSize, "2048");
tempProperties.put(OnsTraceConstants.MaxBatchNum, "100");
tempProperties.put(OnsTraceConstants.NAMESRV_ADDR, this.getNameServerAddr());
tempProperties.put(OnsTraceConstants.InstanceName, "PID_CLIENT_INNER_TRACE_PRODUCER");
tempProperties.put(OnsTraceConstants.TraceDispatcherType, OnsTraceDispatcherType.PRODUCER.name());
AsyncArrayDispatcher dispatcher = new AsyncArrayDispatcher(tempProperties, sessionCredentials);
dispatcher.setHostProducer(defaultMQProducer.getDefaultMQProducerImpl());
traceDispatcher = dispatcher;
this.defaultMQProducer.getDefaultMQProducerImpl().registerSendMessageHook(
new OnsClientSendMessageHookImpl(traceDispatcher));
} catch (Throwable e) {
LOGGER.error("system mqtrace hook init failed ,maybe can't send msg trace data", e);
}
}
}