Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[ISSUE-#4166] Optimize notifyCenter code #4167

Merged
merged 2 commits into from
Nov 9, 2020
Merged

[ISSUE-#4166] Optimize notifyCenter code #4167

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
b054879
enhance notify center code quality
horizonzy Nov 7, 2020
57a5b5d
reverse the judge logic
horizonzy Nov 9, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 21 additions & 23 deletions common/src/main/java/com/alibaba/nacos/common/notify/NotifyCenter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,19 +18,19 @@

import com.alibaba.nacos.api.exception.runtime.NacosRuntimeException;
import com.alibaba.nacos.common.JustForTest;
import com.alibaba.nacos.common.notify.listener.Subscriber;
import com.alibaba.nacos.common.notify.listener.SmartSubscriber;
import com.alibaba.nacos.common.notify.listener.Subscriber;
import com.alibaba.nacos.common.utils.BiFunction;
import com.alibaba.nacos.common.utils.ClassUtils;
import com.alibaba.nacos.common.utils.MapUtils;
import com.alibaba.nacos.common.utils.ThreadUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Iterator;
import java.util.Map;
import java.util.ServiceLoader;
import java.util.NoSuchElementException;
import java.util.Iterator;
import java.util.ServiceLoader;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;

Expand Down Expand Up @@ -170,7 +170,6 @@ public static void shutdown() {
* @param <T> event type
*/
public static <T> void registerSubscriber(final Subscriber consumer) {
final Class<? extends Event> cls = consumer.subscribeType();
// 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) {
Expand All @@ -186,16 +185,17 @@ public static <T> void registerSubscriber(final Subscriber consumer) {
return;
}

if (ClassUtils.isAssignableFrom(SlowEvent.class, cls)) {
INSTANCE.sharePublisher.addSubscriber(consumer, cls);
final Class<? extends Event> subscribeType = consumer.subscribeType();
if (ClassUtils.isAssignableFrom(SlowEvent.class, subscribeType)) {
INSTANCE.sharePublisher.addSubscriber(consumer, subscribeType);
return;
}

addSubscriber(consumer, consumer.subscribeType());
addSubscriber(consumer, subscribeType);
}

/**
* Add a subscriber to pusblisher.
* Add a subscriber to publisher.
*
* @param consumer subscriber instance.
* @param subscribeType subscribeType.
Expand All @@ -217,7 +217,6 @@ private static void addSubscriber(final Subscriber consumer, Class<? extends Eve
* @param consumer subscriber instance.
*/
public static <T> void deregisterSubscriber(final Subscriber consumer) {
final Class<? extends Event> cls = consumer.subscribeType();
if (consumer instanceof SmartSubscriber) {
for (Class<? extends Event> subscribeType : ((SmartSubscriber) consumer).subscribeTypes()) {
if (ClassUtils.isAssignableFrom(SlowEvent.class, subscribeType)) {
Expand All @@ -229,15 +228,16 @@ public static <T> void deregisterSubscriber(final Subscriber consumer) {
return;
}

if (ClassUtils.isAssignableFrom(SlowEvent.class, cls)) {
INSTANCE.sharePublisher.removeSubscriber(consumer, cls);
final Class<? extends Event> subscribeType = consumer.subscribeType();
if (ClassUtils.isAssignableFrom(SlowEvent.class, subscribeType)) {
INSTANCE.sharePublisher.removeSubscriber(consumer, subscribeType);
return;
}

if (removeSubscriber(consumer, consumer.subscribeType())) {
if (removeSubscriber(consumer, subscribeType)) {
return;
}
throw new NoSuchElementException("The subcriber has no event publisher");
throw new NoSuchElementException("The subscriber has no event publisher");
}

/**
Expand All @@ -250,12 +250,11 @@ public static <T> void deregisterSubscriber(final Subscriber consumer) {
private static boolean removeSubscriber(final Subscriber consumer, Class<? extends Event> subscribeType) {

final String topic = ClassUtils.getCanonicalName(subscribeType);
if (INSTANCE.publisherMap.containsKey(topic)) {
EventPublisher publisher = INSTANCE.publisherMap.get(topic);
publisher.removeSubscriber(consumer);
EventPublisher eventPublisher = INSTANCE.publisherMap.get(topic);
horizonzy marked this conversation as resolved.
Show resolved Hide resolved
if (eventPublisher != null) {
eventPublisher.removeSubscriber(consumer);
return true;
}

return false;
}

Expand All @@ -281,16 +280,16 @@ public static boolean publishEvent(final Event event) {
* @param event event instance.
*/
private static boolean publishEvent(final Class<? extends Event> eventType, final Event event) {
final String topic = ClassUtils.getCanonicalName(eventType);
if (ClassUtils.isAssignableFrom(SlowEvent.class, eventType)) {
return INSTANCE.sharePublisher.publish(event);
}

if (INSTANCE.publisherMap.containsKey(topic)) {
EventPublisher publisher = INSTANCE.publisherMap.get(topic);
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;
}
Expand Down Expand Up @@ -321,8 +320,7 @@ public static EventPublisher registerToPublisher(final Class<? extends Event> ev
// MapUtils.computeIfAbsent is a unsafe method.
MapUtils.computeIfAbsent(INSTANCE.publisherMap, topic, publisherFactory, eventType, queueMaxSize);
}
EventPublisher publisher = INSTANCE.publisherMap.get(topic);
return publisher;
return INSTANCE.publisherMap.get(topic);
}

/**
Expand Down