nacos 消息机制
- 目录
-
- 概 述
- 小结
- 参考资料和推荐阅读
LD is tigger forever，CG are not brothers forever， throw the pot and shine forever.
Modesty is not false, solid is not naive, treacherous but not deceitful, stay with good people, and stay away from poor people.
talk is cheap, show others the code and KPI， Keep progress，make a better result.
Survive during the day and develop at night。
目录
概 述
注册中心：
我们讨论一下nacos 的消息机制：我们先考虑一下事件通知：
这里涉及到了Subscriber 的方式：
public abstract class Subscriber<T extends Event> {
/**
* Event callback.
*
* @param event {@link Event}
*/
public abstract void onEvent(T event);
/**
* Type of this subscriber's subscription.
*
* @return Class which extends {@link Event}
*/
public abstract Class<? extends Event> subscribeType();
/**
* It is up to the listener to determine whether the callback is asynchronous or synchronous.
*
* @return {@link Executor}
*/
public Executor executor() {
return null;
}
/**
* Whether to ignore expired events.
*
* @return default value is {@link Boolean#FALSE}
*/
public boolean ignoreExpireEvent() {
return false;
}
}
另外还有一个是Nacoc的事件发布者，源码如下：
public interface EventPublisher extends Closeable {
/**
* Initializes the event publisher.
*
* @param type {@link Event >}
* @param bufferSize Message staging queue size
*/
void init(Class<? extends Event> type, int bufferSize);
/**
* The number of currently staged events.
*
* @return event size
*/
long currentEventSize();
/**
* Add listener.
*
* @param subscriber {@link Subscriber}
*/
void addSubscriber(Subscriber subscriber);
/**
* Remove listener.
*
* @param subscriber {@link Subscriber}
*/
void removeSubscriber(Subscriber subscriber);
/**
* publish event.
*
* @param event {@link Event}
* @return publish event is success
*/
boolean publish(Event event);
/**
* Notify listener.
*
* @param subscriber {@link Subscriber}
* @param event {@link Event}
*/
void notifySubscriber(Subscriber subscriber, Event event);
}
我们从队列内部获取事件并通知订阅者流程如下：
对于单事件发布者：发布者和事件的关系，事件和订阅者的关系，发布者和订阅者的关系是相对比较明确的。
相关的流程图如下：
最后由通知中心完成交互。
nacos就会通过触发NotifyCenter的事件发布方法，事件发布者EventPublisher来对当前事件进行一个广播操作。
涉及的核心类：
源码如下：
public class NotifyCenter {
private static final Logger LOGGER = LoggerFactory.getLogger(NotifyCenter.class);
public static int ringBufferSize;
public static int shareBufferSize;
private static final AtomicBoolean CLOSED = new AtomicBoolean(false);
private static final EventPublisherFactory DEFAULT_PUBLISHER_FACTORY;
private static final NotifyCenter INSTANCE = new NotifyCenter();
private DefaultSharePublisher sharePublisher;
private static Class<? extends EventPublisher> clazz;
/**
* Publisher management container.
*/
private final Map<String, EventPublisher> publisherMap = new ConcurrentHashMap<>(16);
static {
// Internal ArrayBlockingQueue buffer size. For applications with high write throughput,
// this value needs to be increased appropriately. default value is 16384
String ringBufferSizeProperty = "nacos.core.notify.ring-buffer-size";
ringBufferSize = Integer.getInteger(ringBufferSizeProperty, 16384);
// The size of the public publisher's message staging queue buffer
String shareBufferSizeProperty = "nacos.core.notify.share-buffer-size";
shareBufferSize = Integer.getInteger(shareBufferSizeProperty, 1024);
final Collection<EventPublisher> publishers = NacosServiceLoader.load(EventPublisher.class);
Iterator<EventPublisher> iterator = publishers.iterator();
if (iterator.hasNext()) {
clazz = iterator.next().getClass();
} else {
clazz = DefaultPublisher.class;
}
DEFAULT_PUBLISHER_FACTORY = (cls, buffer) -> {
try {
EventPublisher publisher = clazz.newInstance();
publisher.init(cls, buffer);
return publisher;
} catch (Throwable ex) {
LOGGER.error("Service class newInstance has error : ", ex);
throw new NacosRuntimeException(SERVER_ERROR, ex);
}
};
try {
// Create and init DefaultSharePublisher instance.
INSTANCE.sharePublisher = new DefaultSharePublisher();
INSTANCE.sharePublisher.init(SlowEvent.class, shareBufferSize);
} catch (Throwable ex) {
LOGGER.error("Service class newInstance has error : ", ex);
}
ThreadUtils.addShutdownHook(NotifyCenter::shutdown);
}
@JustForTest
public static Map<String, EventPublisher> getPublisherMap() {
return INSTANCE.publisherMap;
}
@JustForTest
public static EventPublisher getPublisher(Class<? extends Event> topic) {
if (ClassUtils.isAssignableFrom(SlowEvent.class, topic)) {
return INSTANCE.sharePublisher;
}
return INSTANCE.publisherMap.get(topic.getCanonicalName());
}
@JustForTest
public static EventPublisher getSharePublisher() {
return INSTANCE.sharePublisher;
}
/**
* Shutdown the several publisher instance which notify center has.
*/
public static void shutdown() {
if (!CLOSED.compareAndSet(false, true)) {
return;
}
LOGGER.warn("[NotifyCenter] Start destroying Publisher");
for (Map.Entry<String, EventPublisher> entry : INSTANCE.publisherMap.entrySet()) {
try {
EventPublisher eventPublisher = entry.getValue();
eventPublisher.shutdown();
} catch (Throwable e) {
LOGGER.error("[EventPublisher] shutdown has error : ", e);
}
}
try {
INSTANCE.sharePublisher.shutdown();
} catch (Throwable e) {
LOGGER.error("[SharePublisher] shutdown has error : ", e);
}
LOGGER.warn("[NotifyCenter] Destruction of the end");
}
/**
* Register a Subscriber. If the Publisher concerned by the Subscriber does not exist, then PublihserMap will
* preempt a placeholder Publisher with default EventPublisherFactory first.
*
* @param consumer subscriber
*/
public static void registerSubscriber(final Subscriber consumer) {
registerSubscriber(consumer, DEFAULT_PUBLISHER_FACTORY);
}
/**
* Register a Subscriber. If the Publisher concerned by the Subscriber does not exist, then PublihserMap will
* preempt a placeholder Publisher with specified EventPublisherFactory first.
*
* @param consumer subscriber
* @param factory publisher factory.
*/
public static void registerSubscriber(final Subscriber consumer, final EventPublisherFactory factory) {
// If you want to listen to multiple events, you do it separately,
// based on subclass's subscribeTypes method return list, it can register to publisher.
if (consumer instanceof SmartSubscriber) {
for (Class<? extends Event> subscribeType : ((SmartSubscriber) consumer).subscribeTypes()) {
// For case, producer: defaultSharePublisher -> consumer: smartSubscriber.
if (ClassUtils.isAssignableFrom(SlowEvent.class, subscribeType)) {
INSTANCE.sharePublisher.addSubscriber(consumer, subscribeType);
} else {
// For case, producer: defaultPublisher -> consumer: subscriber.
addSubscriber(consumer, subscribeType, factory);
}
}
return;
}
final Class<? extends Event> subscribeType = consumer.subscribeType();
if (ClassUtils.isAssignableFrom(SlowEvent.class, subscribeType)) {
INSTANCE.sharePublisher.addSubscriber(consumer, subscribeType);
return;
}
addSubscriber(consumer, subscribeType, factory);
}
/**
* Add a subscriber to publisher.
*
* @param consumer subscriber instance.
* @param subscribeType subscribeType.
* @param factory publisher factory.
*/
private static void addSubscriber(final Subscriber consumer, Class<? extends Event> subscribeType,
EventPublisherFactory factory) {
final String topic = ClassUtils.getCanonicalName(subscribeType);
synchronized (NotifyCenter.class) {
// MapUtils.computeIfAbsent is a unsafe method.
MapUtil.computeIfAbsent(INSTANCE.publisherMap, topic, factory, subscribeType, ringBufferSize);
}
EventPublisher publisher = INSTANCE.publisherMap.get(topic);
if (publisher instanceof ShardedEventPublisher) {
((ShardedEventPublisher) publisher).addSubscriber(consumer, subscribeType);
} else {
publisher.addSubscriber(consumer);
}
}
/**
* Deregister subscriber.
*
* @param consumer subscriber instance.
*/
public static void deregisterSubscriber(final Subscriber consumer) {
if (consumer instanceof SmartSubscriber) {
for (Class<? extends Event> subscribeType : ((SmartSubscriber) consumer).subscribeTypes()) {
if (ClassUtils.isAssignableFrom(SlowEvent.class, subscribeType)) {
INSTANCE.sharePublisher.removeSubscriber(consumer, subscribeType);
} else {
removeSubscriber(consumer, subscribeType);
}
}
return;
}
final Class<? extends Event> subscribeType = consumer.subscribeType();
if (ClassUtils.isAssignableFrom(SlowEvent.class, subscribeType)) {
INSTANCE.sharePublisher.removeSubscriber(consumer, subscribeType);
return;
}
if (removeSubscriber(consumer, subscribeType)) {
return;
}
throw new NoSuchElementException("The subscriber has no event publisher");
}
/**
* Remove subscriber.
*
* @param consumer subscriber instance.
* @param subscribeType subscribeType.
* @return whether remove subscriber successfully or not.
*/
private static boolean removeSubscriber(final Subscriber consumer, Class<? extends Event> subscribeType) {
final String topic = ClassUtils.getCanonicalName(subscribeType);
EventPublisher eventPublisher = INSTANCE.publisherMap.get(topic);
if (null == eventPublisher) {
return false;
}
if (eventPublisher instanceof ShardedEventPublisher) {
((ShardedEventPublisher) eventPublisher).removeSubscriber(consumer, subscribeType);
} else {
eventPublisher.removeSubscriber(consumer);
}
return true;
}
/**
* Request publisher publish event Publishers load lazily, calling publisher. Start () only when the event is
* actually published.
*
* @param event class Instances of the event.
*/
public static boolean publishEvent(final Event event) {
try {
return publishEvent(event.getClass(), event);
} catch (Throwable ex) {
LOGGER.error("There was an exception to the message publishing : ", ex);
return false;
}
}
/**
* Request publisher publish event Publishers load lazily, calling publisher.
*
* @param eventType class Instances type of the event type.
* @param event event instance.
*/
private static boolean publishEvent(final Class<? extends Event> eventType, final Event event) {
if (ClassUtils.isAssignableFrom(SlowEvent.class, eventType)) {
return INSTANCE.sharePublisher.publish(event);
}
final String topic = ClassUtils.getCanonicalName(eventType);
EventPublisher publisher = INSTANCE.publisherMap.get(topic);
if (publisher != null) {
return publisher.publish(event);
}
LOGGER.warn("There are no [{}] publishers for this event, please register", topic);
return false;
}
小结
参考资料和推荐阅读
1.链接: link