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