序
本文主要研究一下AbstractOMSProducer
AbstractOMSProducer
io/openmessaging/rocketmq/producer/AbstractOMSProducer.java
abstract class AbstractOMSProducer implements ServiceLifecycle, MessageFactory { final static Logger log = ClientLogger.getLog(); final KeyValue properties; final DefaultMQProducer rocketmqProducer; private boolean started = false; final ClientConfig clientConfig; AbstractOMSProducer(final KeyValue properties) { this.properties = properties; this.rocketmqProducer = new DefaultMQProducer(); this.clientConfig = BeanUtils.populate(properties, ClientConfig.class); String accessPoints = clientConfig.getOmsAccessPoints(); if (accessPoints == null || accessPoints.isEmpty()) { throw new OMSRuntimeException("-1", "OMS AccessPoints is null or empty."); } this.rocketmqProducer.setNamesrvAddr(accessPoints.replace(',', ';')); this.rocketmqProducer.setProducerGroup(clientConfig.getRmqProducerGroup()); String producerId = buildInstanceName(); this.rocketmqProducer.setSendMsgTimeout(clientConfig.getOmsOperationTimeout()); this.rocketmqProducer.setInstanceName(producerId); this.rocketmqProducer.setMaxMessageSize(1024 * 1024 * 4); properties.put(PropertyKeys.PRODUCER_ID, producerId); } @Override public synchronized void startup() { if (!started) { try { this.rocketmqProducer.start(); } catch (MQClientException e) { throw new OMSRuntimeException("-1", e); } } this.started = true; } @Override public synchronized void shutdown() { if (this.started) { this.rocketmqProducer.shutdown(); } this.started = false; } OMSRuntimeException checkProducerException(String topic, String msgId, Throwable e) { if (e instanceof MQClientException) { if (e.getCause() != null) { if (e.getCause() instanceof RemotingTimeoutException) { return new OMSTimeOutException("-1", String.format("Send message to broker timeout, %dms, Topic=%s, msgId=%s", this.rocketmqProducer.getSendMsgTimeout(), topic, msgId), e); } else if (e.getCause() instanceof MQBrokerException || e.getCause() instanceof RemotingConnectException) { MQBrokerException brokerException = (MQBrokerException) e.getCause(); return new OMSRuntimeException("-1", String.format("Received a broker exception, Topic=%s, msgId=%s, %s", topic, msgId, brokerException.getErrorMessage()), e); } } // Exception thrown by local. else { MQClientException clientException = (MQClientException) e; if (-1 == clientException.getResponseCode()) { return new OMSRuntimeException("-1", String.format("Topic does not exist, Topic=%s, msgId=%s", topic, msgId), e); } else if (ResponseCode.MESSAGE_ILLEGAL == clientException.getResponseCode()) { return new OMSMessageFormatException("-1", String.format("A illegal message for RocketMQ, Topic=%s, msgId=%s", topic, msgId), e); } } } return new OMSRuntimeException("-1", "Send message to RocketMQ broker failed.", e); } protected void checkMessageType(Message message) { if (!(message instanceof BytesMessage)) { throw new OMSNotSupportedException("-1", "Only BytesMessage is supported."); } } @Override public BytesMessage createBytesMessageToTopic(final String topic, final byte[] body) { BytesMessage bytesMessage = new BytesMessageImpl(); bytesMessage.setBody(body); bytesMessage.headers().put(MessageHeader.TOPIC, topic); return bytesMessage; } @Override public BytesMessage createBytesMessageToQueue(final String queue, final byte[] body) { BytesMessage bytesMessage = new BytesMessageImpl(); bytesMessage.setBody(body); bytesMessage.headers().put(MessageHeader.QUEUE, queue); return bytesMessage; }} - AbstractOMSProducer实现了ServiceLifecycle以及MessageFactory
- ServiceLifecycle的startup里头调用DefaultMQProducer的start方法，shutdown里头调用DefaultMQProducer的shutdown方法
- MessageFactory的createBytesMessage的方法主要是返回了BytesMessageImpl
MessageFactory
io/openmessaging/openmessaging-api/0.1.0-alpha/openmessaging-api-0.1.0-alpha-sources.jar!/io/openmessaging/MessageFactory.java
/** * A factory interface for creating {@code Message} objects. * * @author vintagewang@apache.org * @author yukon@apache.org */public interface MessageFactory { /** * Creates a {@code BytesMessage} object. A {@code BytesMessage} object is used to send a message containing a * stream of uninterpreted bytes. * * The returned {@code BytesMessage} object only can be sent to the specified topic. * * @param topic the target topic to send * @param body the body data for a message * @return the created {@code BytesMessage} object * @throws OMSRuntimeException if the OMS provider fails to create this message due to some internal error. */ BytesMessage createBytesMessageToTopic(String topic, byte[] body); /** * Creates a {@code BytesMessage} object. A {@code BytesMessage} object is used to send a message containing a * stream of uninterpreted bytes. *
* The returned {@code BytesMessage} object only can be sent to the specified queue. * * @param queue the target queue to send * @param body the body data for a message * @return the created {@code BytesMessage} object * @throws OMSRuntimeException if the OMS provider fails to create this message due to some internal error. */ BytesMessage createBytesMessageToQueue(String queue, byte[] body);}
- 0.1.0-alpha这个版本区分了topic跟queue，不过在最新版已经移除掉topic，统一为createBytesMessage方法，发送到queue
openmessaging-java/openmessaging-api/src/main/java/io/openmessaging/MessageFactory.java
public interface MessageFactory { /** * Creates a {@code BytesMessage} object. A {@code BytesMessage} object is used to send a message containing a * stream of uninterpreted bytes. * * The returned {@code BytesMessage} object only can be sent to the specified queue. * * @param queue the target queue to send * @param body the body data for a message * @return the created {@code BytesMessage} object * @throws OMSRuntimeException if the OMS provider fails to create this message due to some internal error. */ BytesMessage createBytesMessage(String queue, byte[] body);}
BytesMessageImpl
io/openmessaging/rocketmq/domain/BytesMessageImpl.java
public class BytesMessageImpl implements BytesMessage { private KeyValue headers; private KeyValue properties; private byte[] body; public BytesMessageImpl() { this.headers = OMS.newKeyValue(); this.properties = OMS.newKeyValue(); } @Override public byte[] getBody() { return body; } @Override public BytesMessage setBody(final byte[] body) { this.body = body; return this; } @Override public KeyValue headers() { return headers; } @Override public KeyValue properties() { return properties; } @Override public Message putHeaders(final String key, final int value) { headers.put(key, value); return this; } @Override public Message putHeaders(final String key, final long value) { headers.put(key, value); return this; } @Override public Message putHeaders(final String key, final double value) { headers.put(key, value); return this; } @Override public Message putHeaders(final String key, final String value) { headers.put(key, value); return this; } @Override public Message putProperties(final String key, final int value) { properties.put(key, value); return this; } @Override public Message putProperties(final String key, final long value) { properties.put(key, value); return this; } @Override public Message putProperties(final String key, final double value) { properties.put(key, value); return this; } @Override public Message putProperties(final String key, final String value) { properties.put(key, value); return this; } @Override public String toString() { return ToStringBuilder.reflectionToString(this); }} - 用byte[]作为body实现BytesMessage接口
小结
rocketmq的4.2.0版本的AbstractOMSProducer实现了ServiceLifecycle以及MessageFactory，其实现的open-messaging api的版本为0.1.0-alpha，该版本的MessageFactory里头创建message的方法区分了topic和queue，而在最新的0.3.2-alpha-SNAPSHOT版本，已经移除了topic的概念，统一发送到queue。