微服务----2

微服务----2 MQ

MQ消息可靠性

队列持久化

默认生成的交换机，队列和消息都是持久化的

消费者消息确认

使用auto 目前的配置可能导致消息一直重试

失败重试机制

死信交换机

DelayExchange插件

惰性队列

MQ集群

面试篇

nacos分级存储模型

nacos注册

 @CanDistro
    @PostMapping
    @Secured(parser = NamingResourceParser.class, action = ActionTypes.WRITE)
    public String register(HttpServletRequest request) throws Exception {
        
	// 尝试获取namespaceId
    final String namespaceId = WebUtils
        .optional(request, CommonParams.NAMESPACE_ID, Constants.DEFAULT_NAMESPACE_ID);
    // 尝试获取serviceName，其格式为 group_name@@service_name
    final String serviceName = WebUtils.required(request, CommonParams.SERVICE_NAME);
    NamingUtils.checkServiceNameFormat(serviceName);
	// 解析出实例信息，封装为Instance对象
    final Instance instance = parseInstance(request);
	// 注册实例
    serviceManager.registerInstance(namespaceId, serviceName, instance);
    return "ok";
    }

public void registerInstance(String namespaceId, String serviceName, Instance instance) throws NacosException {
	// 创建一个空的service（如果是第一次来注册实例，要先创建一个空service出来，放入注册表）
    // 此时不包含实例信息
    createEmptyService(namespaceId, serviceName, instance.isEphemeral());
==>
public void createServiceIfAbsent(String namespaceId, String serviceName, boolean local, Cluster cluster)
            throws NacosException {
            //尝试从注册表获取服务
        Service service = getService(namespaceId, serviceName);
        if (service == null) {
            
            Loggers.SRV_LOG.info("creating empty service {}:{}", namespaceId, serviceName);
            service = new Service();
            service.setName(serviceName);
            service.setNamespaceId(namespaceId);
            service.setGroupName(NamingUtils.getGroupName(serviceName));
            // now validate the service. if failed, exception will be thrown
            // 记录服务最后一次更新时间
            service.setLastModifiedMillis(System.currentTimeMillis());
            service.recalculateChecksum();
            if (cluster != null) {
                cluster.setService(service);
                service.getClusterMap().put(cluster.getName(), cluster);
            }
            service.validate();
            // 初始化空服务
            putServiceAndInit(service);
            ==》
				            private void putServiceAndInit(Service service) throws NacosException {
				        // 把服务放入注册表
				        putService(service);
				        service = getService(service.getNamespaceId(), service.getName());
				        // 初始化，健康检测
				        service.init();
				        // 服务状态变更监听
				        consistencyService
				                .listen(KeyBuilder.buildInstanceListKey(service.getNamespaceId(), service.getName(), true), service);
				        consistencyService
				                .listen(KeyBuilder.buildInstanceListKey(service.getNamespaceId(), service.getName(), false), service);
				        Loggers.SRV_LOG.info("[NEW-SERVICE] {}", service.toJson());
				    }
            ==》
            if (!local) {
                addOrReplaceService(service);
            }
        }

==>
    // 拿到创建好的service
    Service service = getService(namespaceId, serviceName);
    // 拿不到则抛异常
    if (service == null) {
        throw new NacosException(NacosException.INVALID_PARAM,
                                 "service not found, namespace: " + namespaceId + ", service: " + serviceName);
    }
    // 添加要注册的实例到service中
    addInstance(namespaceId, serviceName, instance.isEphemeral(), instance);
}

== 》 


public void addInstance(String namespaceId, String serviceName, boolean ephemeral, Instance... ips)
    throws NacosException {
	// 监听服务列表用到的key，服务唯一标识，例如：com.alibaba.nacos.naming.iplist.ephemeral.public##DEFAULT_GROUP@@order-service
    String key = KeyBuilder.buildInstanceListKey(namespaceId, serviceName, ephemeral);
    // 获取服务
    Service service = getService(namespaceId, serviceName);
    // 同步锁，避免并发修改的安全问题，同一个服务可能有多个实例，多个实例只能串行执行
    synchronized (service) {
        // 1）获取要更新的实例列表==》 拷贝注册表中旧的实例列表+新注册的实例 得到最终实例列表
        List instanceList = addIpAddresses(service, ephemeral, ips);
		// 2）封装实例列表到Instances对象
        Instances instances = new Instances();
        instances.setInstanceList(instanceList);
		// 3）完成 注册表更新 以及 Nacos集群的数据同步
        consistencyService.put(key, instances);
       // 这里会判断是临时实例还是永久实例，默认是临时的，永久节点使用强一致性
        ==》
	            @Override
	    public void put(String key, Record value) throws NacosException {
	    	// 更新本地注册表
	        onPut(key, value);
			==>
					 public void onPut(String key, Record value) {
		        // 判断是不是临时实例
		        if (KeyBuilder.matchEphemeralInstanceListKey(key)) {
		            // 封装实例列表到Datum
		            Datum datum = new Datum<>();
		            datum.value = (Instances) value;
		            // 唯一标识
		            datum.key = key;
		            datum.timestamp.incrementAndGet();
		            //serviceId为Key,datum为值 缓存起来
		            dataStore.put(key, datum);
		        }
		        
		        if (!listeners.containsKey(key)) {
		            return;
		        }
		        // 把serviceId和当前 *** 作类型存入notifer
		        notifier.addTask(key, DataOperation.CHANGE);
		        	==>
					        	  public void addTask(String datumKey, DataOperation action) {
			            
			            if (services.containsKey(datumKey) && action == DataOperation.CHANGE) {
			                return;
			            }
			            if (action == DataOperation.CHANGE) {
			                services.put(datumKey, StringUtils.EMPTY);
			            }
             // private BlockingQueue> tasks = new ArrayBlockingQueue<>(1024 * 1024);
             			//把serviceId和事件  放入阻塞队列
			            tasks.offer(Pair.with(datumKey, action));

						==》
						        @Override
						        public void run() {
						            Loggers.DISTRO.info("distro notifier started");
						            
						            for (; ; ) {
						                try {
						                //队列放一个任务，我才更新列表
						                    Pair pair = tasks.take();
						                    handle(pair);
						                } catch (Throwable e) {
						                    Loggers.DISTRO.error("[NACOS-DISTRO] Error while handling notifying task", e);
						                }
						            }
        } 
						==》
			        }
		        	==>
		    }
			==>

	        // If upgrade to 2.0.X, do not sync for v1.
	        if (ApplicationUtils.getBean(UpgradeJudgement.class).isUseGrpcFeatures()) {
	            return;
	        }
	        // 异步，同步给nacos其他节点
	        distroProtocol.sync(new DistroKey(key, KeyBuilder.INSTANCE_LIST_KEY_PREFIX), DataOperation.CHANGE,
	                DistroConfig.getInstance().getSyncDelayMillis());
	                ==》
	                // 通知其他节点
	                        for (Member each : memberManager.allMembersWithoutSelf()) {
			            DistroKey distroKeyWithTarget = new DistroKey(distroKey.getResourceKey(), distroKey.getResourceType(),
			                targetServer);
			        DistroDelayTask distroDelayTask = new DistroDelayTask(distroKeyWithTarget, action, delay);
			        distroTaskEngineHolder.getDelayTaskExecuteEngine().addTask(distroKeyWithTarget, distroDelayTask);
			        if (Loggers.DISTRO.isDebugEnabled()) {
			            Loggers.DISTRO.debug("[DISTRO-SCHEDULE] {} to {}", distroKey, targetServer);
			        }
        }
	                ==>
	    }
        ==》
    }
}