From 39611ea28c38617ffeea09c97e4d16e1fa05174f Mon Sep 17 00:00:00 2001 From: Nortyr <42056534+Nortyr@users.noreply.github.com> Date: Wed, 15 Nov 2023 09:43:13 +0800 Subject: [PATCH] feat:RedisMetadataReport implementation (#13303) * feat:Redis metadataReport implementation * feat:Redis metadataReport implementation * remove error code constant * feat:RedisMetadataReport implementation * fix:sonarlint issue * fix:sonarlint bug * remove unused imports * fix:spotless * add notes --- .../store/redis/RedisMetadataReport.java | 330 ++++++++++++++++++ .../store/redis/RedisMetadataReportTest.java | 103 ++++++ 2 files changed, 433 insertions(+) diff --git a/dubbo-metadata/dubbo-metadata-report-redis/src/main/java/org/apache/dubbo/metadata/store/redis/RedisMetadataReport.java b/dubbo-metadata/dubbo-metadata-report-redis/src/main/java/org/apache/dubbo/metadata/store/redis/RedisMetadataReport.java index 453aee6f48..7c64220faa 100644 --- a/dubbo-metadata/dubbo-metadata-report-redis/src/main/java/org/apache/dubbo/metadata/store/redis/RedisMetadataReport.java +++ b/dubbo-metadata/dubbo-metadata-report-redis/src/main/java/org/apache/dubbo/metadata/store/redis/RedisMetadataReport.java @@ -17,9 +17,17 @@ package org.apache.dubbo.metadata.store.redis; import org.apache.dubbo.common.URL; +import org.apache.dubbo.common.config.configcenter.ConfigItem; import org.apache.dubbo.common.logger.ErrorTypeAwareLogger; import org.apache.dubbo.common.logger.LoggerFactory; +import org.apache.dubbo.common.utils.ConcurrentHashMapUtils; +import org.apache.dubbo.common.utils.ConcurrentHashSet; +import org.apache.dubbo.common.utils.JsonUtils; import org.apache.dubbo.common.utils.StringUtils; +import org.apache.dubbo.metadata.MappingChangedEvent; +import org.apache.dubbo.metadata.MappingListener; +import org.apache.dubbo.metadata.MetadataInfo; +import org.apache.dubbo.metadata.ServiceNameMapping; import org.apache.dubbo.metadata.report.identifier.BaseMetadataIdentifier; import org.apache.dubbo.metadata.report.identifier.KeyTypeEnum; import org.apache.dubbo.metadata.report.identifier.MetadataIdentifier; @@ -33,7 +41,9 @@ import java.util.Arrays; import java.util.Collections; import java.util.HashSet; import java.util.List; +import java.util.Map; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import org.apache.commons.pool2.impl.GenericObjectPoolConfig; import redis.clients.jedis.HostAndPort; @@ -41,12 +51,19 @@ import redis.clients.jedis.Jedis; import redis.clients.jedis.JedisCluster; import redis.clients.jedis.JedisPool; import redis.clients.jedis.JedisPoolConfig; +import redis.clients.jedis.JedisPubSub; +import redis.clients.jedis.Transaction; +import redis.clients.jedis.util.JedisClusterCRC16; import static org.apache.dubbo.common.constants.CommonConstants.CLUSTER_KEY; import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_TIMEOUT; +import static org.apache.dubbo.common.constants.CommonConstants.GROUP_CHAR_SEPARATOR; +import static org.apache.dubbo.common.constants.CommonConstants.QUEUES_KEY; import static org.apache.dubbo.common.constants.CommonConstants.TIMEOUT_KEY; import static org.apache.dubbo.common.constants.LoggerCodeConstants.TRANSPORT_FAILED_RESPONSE; import static org.apache.dubbo.metadata.MetadataConstants.META_DATA_STORE_TAG; +import static org.apache.dubbo.metadata.ServiceNameMapping.DEFAULT_MAPPING_GROUP; +import static org.apache.dubbo.metadata.ServiceNameMapping.getAppNames; /** * RedisMetadataReport @@ -61,11 +78,14 @@ public class RedisMetadataReport extends AbstractMetadataReport { private Set jedisClusterNodes; private int timeout; private String password; + private final String root; + private final ConcurrentHashMap mappingDataListenerMap = new ConcurrentHashMap<>(); public RedisMetadataReport(URL url) { super(url); timeout = url.getParameter(TIMEOUT_KEY, DEFAULT_TIMEOUT); password = url.getPassword(); + this.root = url.getGroup(DEFAULT_ROOT); if (url.getParameter(CLUSTER_KEY, false)) { jedisClusterNodes = new HashSet<>(); List urls = url.getBackupUrls(); @@ -209,4 +229,314 @@ public class RedisMetadataReport extends AbstractMetadataReport { throw new RpcException(msg, e); } } + + /** + * Store class and application names using Redis hashes + * key: default 'dubbo:mapping' + * field: class (serviceInterface) + * value: application_names + * @param serviceInterface field(class) + * @param defaultMappingGroup {@link ServiceNameMapping#DEFAULT_MAPPING_GROUP} + * @param newConfigContent new application_names + * @param ticket previous application_names + * @return + */ + @Override + public boolean registerServiceAppMapping( + String serviceInterface, String defaultMappingGroup, String newConfigContent, Object ticket) { + try { + if (null != ticket && !(ticket instanceof String)) { + throw new IllegalArgumentException("redis publishConfigCas requires stat type ticket"); + } + String pathKey = buildMappingKey(defaultMappingGroup); + + return storeMapping(pathKey, serviceInterface, newConfigContent, (String) ticket); + } catch (Exception e) { + logger.warn(TRANSPORT_FAILED_RESPONSE, "", "", "redis publishConfigCas failed.", e); + return false; + } + } + + private boolean storeMapping(String key, String field, String value, String ticket) { + if (pool != null) { + return storeMappingStandalone(key, field, value, ticket); + } else { + return storeMappingInCluster(key, field, value, ticket); + } + } + + /** + * use 'watch' to implement cas. + * Find information about slot distribution by key. + */ + private boolean storeMappingInCluster(String key, String field, String value, String ticket) { + try (JedisCluster jedisCluster = + new JedisCluster(jedisClusterNodes, timeout, timeout, 2, password, new GenericObjectPoolConfig<>())) { + Jedis jedis = jedisCluster.getConnectionFromSlot(JedisClusterCRC16.getSlot(key)); + jedis.watch(key); + String oldValue = jedis.hget(key, field); + if (null == oldValue || null == ticket || oldValue.equals(ticket)) { + Transaction transaction = jedis.multi(); + transaction.hset(key, field, value); + List result = transaction.exec(); + if (null != result) { + jedisCluster.publish(buildPubSubKey(), field); + return true; + } + } else { + jedis.unwatch(); + } + } catch (Throwable e) { + String msg = "Failed to put " + key + ":" + field + " to redis " + value + ", cause: " + e.getMessage(); + logger.error(TRANSPORT_FAILED_RESPONSE, "", "", msg, e); + throw new RpcException(msg, e); + } + return false; + } + + /** + * use 'watch' to implement cas. + * Find information about slot distribution by key. + */ + private boolean storeMappingStandalone(String key, String field, String value, String ticket) { + try (Jedis jedis = pool.getResource()) { + jedis.watch(key); + String oldValue = jedis.hget(key, field); + if (null == oldValue || null == ticket || oldValue.equals(ticket)) { + Transaction transaction = jedis.multi(); + transaction.hset(key, field, value); + transaction.publish(buildPubSubKey(), field); + List result = transaction.exec(); + return null != result; + } + jedis.unwatch(); + } catch (Throwable e) { + String msg = "Failed to put " + key + ":" + field + " to redis " + value + ", cause: " + e.getMessage(); + logger.error(TRANSPORT_FAILED_RESPONSE, "", "", msg, e); + throw new RpcException(msg, e); + } + return false; + } + + /** + * build mapping key + * @param defaultMappingGroup {@link ServiceNameMapping#DEFAULT_MAPPING_GROUP} + * @return + */ + private String buildMappingKey(String defaultMappingGroup) { + return this.root + GROUP_CHAR_SEPARATOR + defaultMappingGroup; + } + + /** + * build pub/sub key + */ + private String buildPubSubKey() { + return buildMappingKey(DEFAULT_MAPPING_GROUP) + GROUP_CHAR_SEPARATOR + QUEUES_KEY; + } + + /** + * get content and use content to complete cas + * @param serviceKey class + * @param group {@link ServiceNameMapping#DEFAULT_MAPPING_GROUP} + */ + @Override + public ConfigItem getConfigItem(String serviceKey, String group) { + String key = buildMappingKey(group); + String content = getMappingData(key, serviceKey); + + return new ConfigItem(content, content); + } + + /** + * get current application_names + */ + private String getMappingData(String key, String field) { + if (pool != null) { + return getMappingDataStandalone(key, field); + } else { + return getMappingDataInCluster(key, field); + } + } + + private String getMappingDataInCluster(String key, String field) { + try (JedisCluster jedisCluster = + new JedisCluster(jedisClusterNodes, timeout, timeout, 2, password, new GenericObjectPoolConfig<>())) { + return jedisCluster.hget(key, field); + } catch (Throwable e) { + String msg = "Failed to get " + key + ":" + field + " from redis cluster , cause: " + e.getMessage(); + logger.error(TRANSPORT_FAILED_RESPONSE, "", "", msg, e); + throw new RpcException(msg, e); + } + } + + private String getMappingDataStandalone(String key, String field) { + try (Jedis jedis = pool.getResource()) { + return jedis.hget(key, field); + } catch (Throwable e) { + String msg = "Failed to get " + key + ":" + field + " from redis , cause: " + e.getMessage(); + logger.error(TRANSPORT_FAILED_RESPONSE, "", "", msg, e); + throw new RpcException(msg, e); + } + } + + /** + * remove listener. If have no listener,thread will dead + */ + @Override + public void removeServiceAppMappingListener(String serviceKey, MappingListener listener) { + MappingDataListener mappingDataListener = mappingDataListenerMap.get(buildPubSubKey()); + if (null != mappingDataListener) { + NotifySub notifySub = mappingDataListener.getNotifySub(); + notifySub.removeListener(serviceKey, listener); + if (notifySub.isEmpty()) { + mappingDataListener.shutdown(); + } + } + } + + /** + * Start a thread and subscribe to {@link this#buildPubSubKey()}. + * Notify {@link MappingListener} if there is a change in the 'application_names' message. + */ + @Override + public Set getServiceAppMapping(String serviceKey, MappingListener listener, URL url) { + MappingDataListener mappingDataListener = + ConcurrentHashMapUtils.computeIfAbsent(mappingDataListenerMap, buildPubSubKey(), k -> { + MappingDataListener dataListener = new MappingDataListener(buildPubSubKey()); + dataListener.start(); + return dataListener; + }); + mappingDataListener.getNotifySub().addListener(serviceKey, listener); + return this.getServiceAppMapping(serviceKey, url); + } + + @Override + public Set getServiceAppMapping(String serviceKey, URL url) { + String key = buildMappingKey(DEFAULT_MAPPING_GROUP); + return getAppNames(getMappingData(key, serviceKey)); + } + + @Override + public MetadataInfo getAppMetadata(SubscriberMetadataIdentifier identifier, Map instanceMetadata) { + String content = this.getMetadata(identifier); + return JsonUtils.toJavaObject(content, MetadataInfo.class); + } + + @Override + public void publishAppMetadata(SubscriberMetadataIdentifier identifier, MetadataInfo metadataInfo) { + this.storeMetadata(identifier, metadataInfo.getContent()); + } + + @Override + public void unPublishAppMetadata(SubscriberMetadataIdentifier identifier, MetadataInfo metadataInfo) { + this.deleteMetadata(identifier); + } + + // for test + public MappingDataListener getMappingDataListener() { + return mappingDataListenerMap.get(buildPubSubKey()); + } + + /** + * Listen for changes in the 'application_names' message and notify the listener. + */ + class NotifySub extends JedisPubSub { + + private final Map> listeners = new ConcurrentHashMap<>(); + + public void addListener(String key, MappingListener listener) { + Set listenerSet = listeners.computeIfAbsent(key, k -> new ConcurrentHashSet<>()); + listenerSet.add(listener); + } + + public void removeListener(String serviceKey, MappingListener listener) { + Set listenerSet = this.listeners.get(serviceKey); + if (listenerSet != null) { + listenerSet.remove(listener); + if (listenerSet.isEmpty()) { + this.listeners.remove(serviceKey); + } + } + } + + public Boolean isEmpty() { + return this.listeners.isEmpty(); + } + + @Override + public void onMessage(String key, String msg) { + logger.info("sub from redis:" + key + " message:" + msg); + String applicationNames = getMappingData(buildMappingKey(DEFAULT_MAPPING_GROUP), msg); + MappingChangedEvent mappingChangedEvent = new MappingChangedEvent(msg, getAppNames(applicationNames)); + if (!listeners.get(msg).isEmpty()) { + for (MappingListener mappingListener : listeners.get(msg)) { + mappingListener.onEvent(mappingChangedEvent); + } + } + } + + @Override + public void onPMessage(String pattern, String key, String msg) { + onMessage(key, msg); + } + + @Override + public void onPSubscribe(String pattern, int subscribedChannels) { + super.onPSubscribe(pattern, subscribedChannels); + } + } + + /** + * Subscribe application names change message. + */ + class MappingDataListener extends Thread { + + private String path; + + private final NotifySub notifySub = new NotifySub(); + // for test + protected volatile boolean running = true; + + public MappingDataListener(String path) { + this.path = path; + } + + public NotifySub getNotifySub() { + return notifySub; + } + + @Override + public void run() { + while (running) { + if (pool != null) { + try (Jedis jedis = pool.getResource()) { + jedis.subscribe(notifySub, path); + } catch (Throwable e) { + String msg = "Failed to subscribe " + path + ", cause: " + e.getMessage(); + logger.error(TRANSPORT_FAILED_RESPONSE, "", "", msg, e); + throw new RpcException(msg, e); + } + } else { + try (JedisCluster jedisCluster = new JedisCluster( + jedisClusterNodes, timeout, timeout, 2, password, new GenericObjectPoolConfig<>())) { + jedisCluster.subscribe(notifySub, path); + } catch (Throwable e) { + String msg = "Failed to subscribe " + path + ", cause: " + e.getMessage(); + logger.error(TRANSPORT_FAILED_RESPONSE, "", "", msg, e); + throw new RpcException(msg, e); + } + } + } + } + + public void shutdown() { + try { + running = false; + notifySub.unsubscribe(path); + } catch (Throwable e) { + String msg = "Failed to unsubscribe " + path + ", cause: " + e.getMessage(); + logger.error(TRANSPORT_FAILED_RESPONSE, "", "", msg, e); + } + } + } } diff --git a/dubbo-metadata/dubbo-metadata-report-redis/src/test/java/org/apache/dubbo/metadata/store/redis/RedisMetadataReportTest.java b/dubbo-metadata/dubbo-metadata-report-redis/src/test/java/org/apache/dubbo/metadata/store/redis/RedisMetadataReportTest.java index ebfce5de95..77646498a8 100644 --- a/dubbo-metadata/dubbo-metadata-report-redis/src/test/java/org/apache/dubbo/metadata/store/redis/RedisMetadataReportTest.java +++ b/dubbo-metadata/dubbo-metadata-report-redis/src/test/java/org/apache/dubbo/metadata/store/redis/RedisMetadataReportTest.java @@ -17,18 +17,26 @@ package org.apache.dubbo.metadata.store.redis; import org.apache.dubbo.common.URL; +import org.apache.dubbo.common.config.configcenter.ConfigItem; import org.apache.dubbo.common.utils.JsonUtils; import org.apache.dubbo.common.utils.NetUtils; +import org.apache.dubbo.metadata.MappingChangedEvent; +import org.apache.dubbo.metadata.MappingListener; +import org.apache.dubbo.metadata.MetadataInfo; import org.apache.dubbo.metadata.definition.ServiceDefinitionBuilder; import org.apache.dubbo.metadata.definition.model.FullServiceDefinition; import org.apache.dubbo.metadata.report.identifier.KeyTypeEnum; import org.apache.dubbo.metadata.report.identifier.MetadataIdentifier; +import org.apache.dubbo.metadata.report.identifier.SubscriberMetadataIdentifier; import org.apache.dubbo.rpc.RpcException; import java.io.IOException; +import java.util.Collections; import java.util.HashMap; import java.util.Map; import java.util.Random; +import java.util.Set; +import java.util.concurrent.CountDownLatch; import org.apache.commons.lang3.SystemUtils; import org.junit.jupiter.api.AfterEach; @@ -44,6 +52,7 @@ import redis.embedded.RedisServer; import static org.apache.dubbo.common.constants.CommonConstants.CONSUMER_SIDE; import static org.apache.dubbo.common.constants.CommonConstants.PROVIDER_SIDE; import static org.apache.dubbo.common.constants.CommonConstants.SYNC_REPORT_KEY; +import static org.apache.dubbo.metadata.ServiceNameMapping.DEFAULT_MAPPING_GROUP; import static redis.embedded.RedisServer.newRedisServer; class RedisMetadataReportTest { @@ -256,4 +265,98 @@ class RedisMetadataReportTest { } } } + + @Test + void testRegisterServiceAppMapping() throws InterruptedException { + String serviceKey1 = "org.apache.dubbo.metadata.store.redis.RedisMetadata4TstService"; + String serviceKey2 = "org.apache.dubbo.metadata.store.redis.RedisMetadata4TstService2"; + + String appNames1 = "test1"; + String appNames2 = "test1,test2"; + CountDownLatch latch = new CountDownLatch(2); + CountDownLatch latch2 = new CountDownLatch(2); + + MappingListener mappingListener = new MappingListener() { + @Override + public void onEvent(MappingChangedEvent event) { + Set apps = event.getApps(); + if (apps.size() == 1) { + Assertions.assertTrue(apps.contains("test1")); + } else { + Assertions.assertTrue(apps.contains("test1")); + Assertions.assertTrue(apps.contains("test2")); + } + if (serviceKey1.equals(event.getServiceKey())) { + latch.countDown(); + } else if (serviceKey2.equals(event.getServiceKey())) { + latch2.countDown(); + } + } + + @Override + public void stop() {} + }; + + Set serviceAppMapping = + redisMetadataReport.getServiceAppMapping(serviceKey1, mappingListener, registryUrl); + + Assertions.assertTrue(serviceAppMapping.isEmpty()); + + ConfigItem configItem = redisMetadataReport.getConfigItem(serviceKey1, DEFAULT_MAPPING_GROUP); + + redisMetadataReport.registerServiceAppMapping( + serviceKey1, DEFAULT_MAPPING_GROUP, appNames1, configItem.getTicket()); + configItem = redisMetadataReport.getConfigItem(serviceKey1, DEFAULT_MAPPING_GROUP); + + redisMetadataReport.registerServiceAppMapping( + serviceKey1, DEFAULT_MAPPING_GROUP, appNames2, configItem.getTicket()); + + latch.await(); + + serviceAppMapping = redisMetadataReport.getServiceAppMapping(serviceKey2, mappingListener, registryUrl); + + Assertions.assertTrue(serviceAppMapping.isEmpty()); + + configItem = redisMetadataReport.getConfigItem(serviceKey2, DEFAULT_MAPPING_GROUP); + + redisMetadataReport.registerServiceAppMapping( + serviceKey2, DEFAULT_MAPPING_GROUP, appNames1, configItem.getTicket()); + configItem = redisMetadataReport.getConfigItem(serviceKey2, DEFAULT_MAPPING_GROUP); + redisMetadataReport.registerServiceAppMapping( + serviceKey2, DEFAULT_MAPPING_GROUP, appNames2, configItem.getTicket()); + + latch2.await(); + RedisMetadataReport.MappingDataListener mappingDataListener = redisMetadataReport.getMappingDataListener(); + Assertions.assertTrue(mappingDataListener.running); + Assertions.assertTrue(!mappingDataListener.getNotifySub().isEmpty()); + + redisMetadataReport.removeServiceAppMappingListener(serviceKey1, mappingListener); + Assertions.assertTrue(mappingDataListener.running); + Assertions.assertTrue(!mappingDataListener.getNotifySub().isEmpty()); + redisMetadataReport.removeServiceAppMappingListener(serviceKey2, mappingListener); + Assertions.assertTrue(!mappingDataListener.running); + Assertions.assertTrue(mappingDataListener.getNotifySub().isEmpty()); + } + + @Test + void testAppMetadata() { + String serviceKey = "org.apache.dubbo.metadata.store.redis.RedisMetadata4TstService"; + String appName = "demo"; + URL url = URL.valueOf("test://127.0.0.1:8888/" + serviceKey); + + MetadataInfo metadataInfo = new MetadataInfo(appName); + metadataInfo.addService(url); + SubscriberMetadataIdentifier identifier = + new SubscriberMetadataIdentifier(appName, metadataInfo.calAndGetRevision()); + MetadataInfo appMetadata = redisMetadataReport.getAppMetadata(identifier, Collections.emptyMap()); + Assertions.assertNull(appMetadata); + + redisMetadataReport.publishAppMetadata(identifier, metadataInfo); + appMetadata = redisMetadataReport.getAppMetadata(identifier, Collections.emptyMap()); + Assertions.assertNotNull(appMetadata); + Assertions.assertEquals(appMetadata.toFullString(), metadataInfo.toFullString()); + redisMetadataReport.unPublishAppMetadata(identifier, metadataInfo); + appMetadata = redisMetadataReport.getAppMetadata(identifier, Collections.emptyMap()); + Assertions.assertNull(appMetadata); + } }