MQ入门之看MQ项目源码 XXL-MQ

类图架构 broker模块

provider/consumer

样例

无spring样例

兼容spring样例

从示范类深入使用流程

从frameless框架入手，有四个文件

XxlMqFramelessApplication

这是实例入口类，我们可以看到，先是使用了getInstance()再使用start()，之后每小时检查一次是否被中断，有就关闭类，这个getInstance()实际上是取了一个XxlMqConf单例，这将在下一部分展开

	public class XxlMqFramelessApplication {
	    public static void main(String[] args) throws Exception {
	        // consumer list
	        List consumerList = new ArrayList<>();
	        consumerList.add(new Demo2MqComsumer());
	        // 开启消费者
	        XxlMqConf.getInstance().start(consumerList);
	        // 开启生产者
	        XxlMqProducer.produce(new XxlMqMessage("topic_2","test msg data"));
	        //每小时检查一次是否被中断，有就关闭
	        while (!Thread.currentThread().isInterrupted()) {
	            TimeUnit.HOURS.sleep(1);
	        }
	        // stop
	        XxlMqConf.getInstance().stop();
	    }
	}

Demo2MqComsumer

	@MqConsumer(topic = "topic_2")
	public class Demo2MqComsumer implements IMqConsumer {
	    private Logger logger = LoggerFactory.getLogger(Demo2MqComsumer.class);
	    @Override
	    public MqResult consume(String data) throws Exception {
	        logger.info("[Demo2MqComsumer] 消费一条消息:{}", data);
	        return MqResult.SUCCESS;
	    }
	}

XxlMqConf

主要分为三个部分

静态加载

Conf类采用了饿汉式的单例模式来新建Conf实例，并使用factory生成一个self4j日志。

	private static Logger logger = LoggerFactory.getLogger(XxlMqConf.class);

    private static XxlMqConf instance = new XxlMqConf();
    public static XxlMqConf getInstance() {
        return instance;
    }

    // XxlMqClientFactory
    private XxlMqClientFactory xxlMqClientFactory;

读取配置文件并初始化配置

	public void start(List consumerList){

	     // 取出properties文件
	     Properties xxlJobProp = loadProperties("xxl-mq.properties");
	
	     xxlMqClientFactory = new XxlMqClientFactory();
	     xxlMqClientFactory.setAdminAddress(xxlJobProp.getProperty("xxl.mq.admin.address"));
	     xxlMqClientFactory.setAccessToken(xxlJobProp.getProperty("xxl.mq.accessToken"));
	     xxlMqClientFactory.setConsumerList(consumerList);
	     xxlMqClientFactory.init();
     }
	
	public static Properties loadProperties(String propertyFileName) {
        InputStreamReader in = null;
        try {
            ClassLoader loder = Thread.currentThread().getContextClassLoader();

            in = new InputStreamReader(loder.getResourceAsStream(propertyFileName), "UTF-8");;
            if (in != null) {
                Properties prop = new Properties();
                prop.load(in);
                return prop;
            }
        } catch (IOException e) {
            logger.error("load {} error!", propertyFileName);
        } finally {
            if (in != null) {
                try {
                    in.close();
                } catch (IOException e) {
                    logger.error("close {} error!", propertyFileName);
                }
            }
        }
        return null;
    }

终止部分

	public void stop() throws Exception {
		if (xxlMqClientFactory != null) {
		    xxlMqClientFactory.destroy();
		}
	}

核心类之XxlMqClientFactory init()

前情提要，XxlMqConf.getInstance().start(consumerList)，从XxlMqConf中可以看出，已经拥有了单例Conf，然后start()中初始化了一个XxlMqClientFactory并且调用了其init()，以下是init()源码

    public void init() {
        //判断是否合法
        validConsumer();
        // 开启转发存储服务
        startBrokerService();
        // 开启消费者
        startConsumer();
    }

startBrokerService()

部分1初始化了XxlRpcInvokerFactory，然后会在底层使用HttpURLConnection向注册中心和发现中心传输
部分2写了一个Netty客户端，并且写了一个代理完成远程RPC
部分3开启了6个线程，3个负责直接传输，3个负责传输后会回调的任务,这六个线程会从阻塞队列中取出任务

public void startBrokerService() {

        // 部分1
        xxlRpcInvokerFactory = new XxlRpcInvokerFactory(XxlRegistryServiceRegistry.class, new HashMap(){{
            put(XxlRegistryServiceRegistry.XXL_REGISTRY_ADDRESS, adminAddress);
            put(XxlRegistryServiceRegistry.ACCESS_TOKEN, accessToken);
        }});
        try {
            xxlRpcInvokerFactory.start();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        // init ConsumerRegistryHelper
        XxlRegistryServiceRegistry commonServiceRegistry = (XxlRegistryServiceRegistry) xxlRpcInvokerFactory.getServiceRegistry();
        consumerRegistryHelper = new ConsumerRegistryHelper(commonServiceRegistry);


        // 部分2
        xxlMqBroker = (IXxlMqBroker) new XxlRpcReferenceBean(
                NetEnum.NETTY,
                Serializer.SerializeEnum.HESSIAN.getSerializer(),
                CallType.SYNC,
                LoadBalance.ROUND,
                IXxlMqBroker.class,
                null,
                10000,
                null,
                null,
                null,
                xxlRpcInvokerFactory).getObject();
        // async + mult, addMessages
        for (int i = 0; i < 3; i++) {
            clientFactoryThreadPool.execute(new Runnable() {
                @Override
                public void run() {
                    while (!XxlMqClientFactory.clientFactoryPoolStoped) {
                        try {
                            XxlMqMessage message = newMessageQueue.take();
                            if (message != null) {
                                // load
                                List messageList = new ArrayList<>();
                                messageList.add(message);
                                List otherMessageList = new ArrayList<>();
                                int drainToNum = newMessageQueue.drainTo(otherMessageList, 100);
                                if (drainToNum > 0) {
                                    messageList.addAll(otherMessageList);
                                }
                                // save
                                xxlMqBroker.addMessages(messageList);
                            }
                        } catch (Exception e) {
                            if (!XxlMqClientFactory.clientFactoryPoolStoped) {
                                logger.error(e.getMessage(), e);
                            }
                        }
                    }
                    // finally total
                    List otherMessageList = new ArrayList<>();
                    int drainToNum = newMessageQueue.drainTo(otherMessageList);
                    if (drainToNum> 0) {
                        xxlMqBroker.addMessages(otherMessageList);
                    }
                }
            });
        }


		//部分3
        // async + mult, addMessages
        for (int i = 0; i < 3; i++) {
            clientFactoryThreadPool.execute(new Runnable() {
                @Override
                public void run() {
                    while (!XxlMqClientFactory.clientFactoryPoolStoped) {
                        try {
                            XxlMqMessage message = callbackMessageQueue.take();
                            if (message != null) {
                                // load
                                List messageList = new ArrayList<>();
                                messageList.add(message);
                                List otherMessageList = new ArrayList<>();
                                int drainToNum = callbackMessageQueue.drainTo(otherMessageList, 100);
                                if (drainToNum > 0) {
                                    messageList.addAll(otherMessageList);
                                }
                                // callback
                                xxlMqBroker.callbackMessages(messageList);
                            }
                        } catch (Exception e) {
                            if (!XxlMqClientFactory.clientFactoryPoolStoped) {
                                logger.error(e.getMessage(), e);
                            }
                        }
                    }
                    // finally total
                    List otherMessageList = new ArrayList<>();
                    int drainToNum = callbackMessageQueue.drainTo(otherMessageList);
                    if (drainToNum> 0) {
                        xxlMqBroker.callbackMessages(otherMessageList);
                    }
                }
            });
        }
    }

部分1深入查看

// init XxlRpcInvokerFactory
        xxlRpcInvokerFactory = new XxlRpcInvokerFactory(XxlRegistryServiceRegistry.class, new HashMap(){{
            put(XxlRegistryServiceRegistry.XXL_REGISTRY_ADDRESS, adminAddress);
            put(XxlRegistryServiceRegistry.ACCESS_TOKEN, accessToken);
        }});
        try {
            xxlRpcInvokerFactory.start();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

以下是xxlRpcInvokerFactory.start()，先新建了一个服务注册类，然后

    public void start() throws Exception {
        // start registry
        if (serviceRegistryClass != null) {
            serviceRegistry = serviceRegistryClass.newInstance();
            serviceRegistry.start(serviceRegistryParam);
        }
    }

以下是xxlRpcInvokerFactory.start()，

    public void start(Map param) {
        String xxlRegistryAddress = param.get(XXL_REGISTRY_ADDRESS);
        String accessToken = param.get(ACCESS_TOKEN);
        String biz = param.get(BIZ);
        String env = param.get(ENV);

        // fill
        biz = (biz!=null&&biz.trim().length()>0)?biz:"default";
        env = (env!=null&&env.trim().length()>0)?env:"default";

        xxlRegistryClient = new XxlRegistryClient(xxlRegistryAddress, accessToken, biz, env);
    }

下面这个方法就是新建了一个XxlRegistryBaseClient类，并开启了注册和发现两个守护进程

    public XxlRegistryClient(String adminAddress, String accessToken, String biz, String env) {
        registryBaseClient = new XxlRegistryBaseClient(adminAddress, accessToken, biz, env);
        // registry thread
        registryThread = new Thread(new Runnable() {
            @Override
            public void run() {
                while (!registryThreadStop) {
                    try {
                        if (registryData.size() > 0) {
                            boolean ret = registryBaseClient.registry(new ArrayList(registryData));
                        }
                    } catch (Exception e) {
                        if (!registryThreadStop) {
                            logger.error(">>>>>>>>>>> xxl-registry, registryThread error.", e);
                        }
                    }
                    try {
                        TimeUnit.SECONDS.sleep(10);
                    } catch (Exception e) {
                        if (!registryThreadStop) {
                            logger.error(">>>>>>>>>>> xxl-registry, registryThread error.", e);
                        }
                    }
                }
            }
        });
        registryThread.setName("xxl-registry, XxlRegistryClient registryThread.");
        registryThread.setDaemon(true);
        registryThread.start();
        // discovery thread
        discoveryThread = new Thread(new Runnable() {
            @Override
            public void run() {
                while (!registryThreadStop) {
                    if (discoveryData.size() == 0) {
                        try {
                            TimeUnit.SECONDS.sleep(3);
                        } catch (Exception e) {
                            if (!registryThreadStop) {
                                logger.error(">>>>>>>>>>> xxl-registry, discoveryThread error.", e);
                            }
                        }
                    } else {
                        try {
                            // monitor
                            boolean monitorRet = registryBaseClient.monitor(discoveryData.keySet());
                            // avoid fail-retry request too quick
                            if (!monitorRet){
                                TimeUnit.SECONDS.sleep(10);
                            }
                            // refreshDiscoveryData, all
                            refreshDiscoveryData(discoveryData.keySet());
                        } catch (Exception e) {
                            if (!registryThreadStop) {
                                logger.error(">>>>>>>>>>> xxl-registry, discoveryThread error.", e);
                            }
                        }
                    }
                }
            }
        });
        discoveryThread.setName("xxl-registry, XxlRegistryClient discoveryThread.");
        discoveryThread.setDaemon(true);
        discoveryThread.start();
  	}

	public XxlRegistryBaseClient(String adminAddress, String accessToken, String biz, String env) {
        this.adminAddress = adminAddress;
        this.accessToken = accessToken;
        this.biz = biz;
        this.env = env;
        // valid
        if (adminAddress==null || adminAddress.trim().length()==0) {
            throw new RuntimeException("xxl-registry adminAddress empty");
        }
        if (biz==null || biz.trim().length()<4 || biz.trim().length()>255) {
            throw new RuntimeException("xxl-registry biz empty Invalid[4~255]");
        }
        if (env==null || env.trim().length()<2 || env.trim().length()>255) {
            throw new RuntimeException("xxl-registry biz env Invalid[2~255]");
        }
        // parse
        adminAddressArr = new ArrayList<>();
        if (adminAddress.contains(",")) {
            adminAddressArr.addAll(Arrays.asList(adminAddress.split(",")));
        } else {
            adminAddressArr.add(adminAddress);
        }
    }

BasicHttpUtil

这是使用的通信类，用于注册和发现中心通信

	public class BasicHttpUtil {
	    private static Logger logger = LoggerFactory.getLogger(BasicHttpUtil.class);
	    
	    public static String postBody(String url, String requestBody, int timeout) {
	        HttpURLConnection connection = null;
	        BufferedReader bufferedReader = null;
	        try {
	            // connection
	            URL realUrl = new URL(url);
	            connection = (HttpURLConnection) realUrl.openConnection();
	            // connection setting
	            connection.setRequestMethod("POST");
	            connection.setDoOutput(true);
	            connection.setDoInput(true);
	            connection.setUseCaches(false);
	            connection.setReadTimeout(timeout * 1000);
	            connection.setConnectTimeout(3 * 1000);
	            connection.setRequestProperty("connection", "Keep-Alive");
	            connection.setRequestProperty("Content-Type", "application/json;charset=UTF-8");
	            connection.setRequestProperty("Accept-Charset", "application/json;charset=UTF-8");
	            // do connection
	            connection.connect();
	            // write requestBody
	            DataOutputStream dataOutputStream = new DataOutputStream(connection.getOutputStream());
	            dataOutputStream.writeBytes(requestBody);
	            dataOutputStream.flush();
	            dataOutputStream//ppp
	            return result.toString();
	        } catch (Exception e) {
	            logger.error(e.getMessage(), e);
	        } finally {
	            try {
	                if (bufferedReader != null) {
	                    bufferedReader.close();
	                }
	                if (connection != null) {
	                    connection.disconnect();
	                }
	            } catch (Exception e2) {
	                logger.error(e2.getMessage(), e2);
	            }
	        }
	        return null;
	    }
	}

getObject()完成代理类逻辑

这个方法是用于远程发出TCP请求，向broker发送数据

public Object getObject() {
	return Proxy.newProxyInstance(Thread.currentThread()
	.getContextClassLoader(), new Class[] { iface },
	new InvocationHandler() {
		@Override
		public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
	
			// method param
			String className = mjethod.getDeclaringClass().getName();	// iface.getName()
			String varsion_ = version;
			String methodName = method.getName();
			Class[] parameterTypes = method.getParameterTypes();
			Object[] parameters = args;
	
			// filter for generic
			if (className.equals(XxlRpcGenericService.class.getName()) && methodName.equals("invoke")) {
	
				Class[] paramTypes = null;
				if (args[3]!=null) {
					String[] paramTypes_str = (String[]) args[3];
					if (paramTypes_str.length > 0) {
						paramTypes = new Class[paramTypes_str.length];
						for (int i = 0; i < paramTypes_str.length; i++) {
							paramTypes[i] = ClassUtil.resolveClass(paramTypes_str[i]);
						}
					}
				}
	
				className = (String) args[0];
				varsion_ = (String) args[1];
				methodName = (String) args[2];
				parameterTypes = paramTypes;
				parameters = (Object[]) args[4];
			}
	
			// filter method like "Object.toString()"
			if (className.equals(Object.class.getName())) {
				logger.info(">>>>>>>>>>> xxl-rpc proxy class-method not support [{}#{}]", className, methodName);
				throw new XxlRpcException("xxl-rpc proxy class-method not support");
			}
	
			// address
			String finalAddress = address;
			if (finalAddress==null || finalAddress.trim().length()==0) {
				if (invokerFactory!=null && invokerFactory.getServiceRegistry()!=null) {
					// discovery
					String serviceKey = XxlRpcProviderFactory.makeServiceKey(className, varsion_);
					TreeSet addressSet = invokerFactory.getServiceRegistry().discovery(serviceKey);
					// load balance
					if (addressSet==null || addressSet.size()==0) {
						// pass
					} else if (addressSet.size()==1) {
						finalAddress = addressSet.first();
					} else {
						finalAddress = loadBalance.xxlRpcInvokerRouter.route(serviceKey, addressSet);
					}
	
				}
			}
			if (finalAddress==null || finalAddress.trim().length()==0) {
				throw new XxlRpcException("xxl-rpc reference bean["+ className +"] address empty");
			}
	
			// request
			XxlRpcRequest xxlRpcRequest = new XxlRpcRequest();
	                 xxlRpcRequest.setRequestId(UUID.randomUUID().toString());
	                 xxlRpcRequest.setCreateMillisTime(System.currentTimeMillis());
	                 xxlRpcRequest.setAccessToken(accessToken);
	                 xxlRpcRequest.setClassName(className);
	                 xxlRpcRequest.setMethodName(methodName);
	                 xxlRpcRequest.setParameterTypes(parameterTypes);
	                 xxlRpcRequest.setParameters(parameters);
	                 
	                 // send
			if (CallType.SYNC == callType) {
				// future-response set
				XxlRpcFutureResponse futureResponse = new XxlRpcFutureResponse(invokerFactory, xxlRpcRequest, null);
				try {
					// do invoke
					client.asyncSend(finalAddress, xxlRpcRequest);
	
					// future get
					XxlRpcResponse xxlRpcResponse = futureResponse.get(timeout, TimeUnit.MILLISECONDS);
					if (xxlRpcResponse.getErrorMsg() != null) {
						throw new XxlRpcException(xxlRpcResponse.getErrorMsg());
					}
					return xxlRpcResponse.getResult();
				} catch (Exception e) {
					logger.info(">>>>>>>>>>> xxl-rpc, invoke error, address:{}, XxlRpcRequest{}", finalAddress, xxlRpcRequest);
	
					throw (e instanceof XxlRpcException)?e:new XxlRpcException(e);
				} finally{
					// future-response remove
					futureResponse.removeInvokerFuture();
				}
			} else if (CallType.FUTURE == callType) {
				// future-response set
				XxlRpcFutureResponse futureResponse = new XxlRpcFutureResponse(invokerFactory, xxlRpcRequest, null);
	                        try {
					// invoke future set
					XxlRpcInvokeFuture invokeFuture = new XxlRpcInvokeFuture(futureResponse);
					XxlRpcInvokeFuture.setFuture(invokeFuture);
	
	                            // do invoke
	                            client.asyncSend(finalAddress, xxlRpcRequest);
	
	                            return null;
	                        } catch (Exception e) {
					logger.info(">>>>>>>>>>> xxl-rpc, invoke error, address:{}, XxlRpcRequest{}", finalAddress, xxlRpcRequest);
	
					// future-response remove
					futureResponse.removeInvokerFuture();
	
					throw (e instanceof XxlRpcException)?e:new XxlRpcException(e);
	                        }
	
			} else if (CallType.CALLBACK == callType) {
	
				// get callback
				XxlRpcInvokeCallback finalInvokeCallback = invokeCallback;
				XxlRpcInvokeCallback threadInvokeCallback = XxlRpcInvokeCallback.getCallback();
				if (threadInvokeCallback != null) {
					finalInvokeCallback = threadInvokeCallback;
				}
				if (finalInvokeCallback == null) {
					throw new XxlRpcException("xxl-rpc XxlRpcInvokeCallback（CallType="+ CallType.CALLBACK.name() +"） cannot be null.");
				}
	
				// future-response set
				XxlRpcFutureResponse futureResponse = new XxlRpcFutureResponse(invokerFactory, xxlRpcRequest, finalInvokeCallback);
				try {
					client.asyncSend(finalAddress, xxlRpcRequest);
				} catch (Exception e) {
					logger.info(">>>>>>>>>>> xxl-rpc, invoke error, address:{}, XxlRpcRequest{}", finalAddress, xxlRpcRequest);
	
					// future-response remove
					futureResponse.removeInvokerFuture();
	
					throw (e instanceof XxlRpcException)?e:new XxlRpcException(e);
				}
	
				return null;
			} else if (CallType.ONEWAY == callType) {
	                        client.asyncSend(finalAddress, xxlRpcRequest);
	                        return null;
	                    } else {
				throw new XxlRpcException("xxl-rpc callType["+ callType +"] invalid");
			}
		}
	});
}

startConsumer()

一开始，将consumerRespository中合格的消费者(由init()中validConsumer()来判定)全部注册入ConsumerRegistryHelper(由startBrokerService()来创建)，然后正式执行所有消费者

	private void startConsumer() {
        // valid
        if (consumerRespository ==null || consumerRespository.size()==0) {
            return;
        }
        // registry consumer
        getConsumerRegistryHelper().registerConsumer(consumerRespository);
        // execute thread
        for (ConsumerThread item: consumerRespository) {
            clientFactoryThreadPool.execute(item);
            logger.info(">>>>>>>>>>> xxl-mq, consumer init success, , topic:{}, group:{}", item.getMqConsumer().topic(), item.getMqConsumer().group());
        }

    }

将线程所有主题，组和线程标识位取出，注册进入和发现这些消费者

	public void registerConsumer(List consumerThreadList) {
        List registryParamList = new ArrayList<>();
        Set registryParamKeyList = new HashSet<>();
        for (ConsumerThread consumerThread: consumerThreadList) {
            String registryKey = makeRegistryKey(consumerThread.getMqConsumer().topic());
            String registryVal = makeRegistryVal(consumerThread.getMqConsumer().group(), consumerThread.getUuid());
            registryParamList.add(new XxlRegistryDataParamVO(registryKey, registryVal));
            registryParamKeyList.add(registryKey);
        }
        // registry mult consumer
        serviceRegistry.getXxlRegistryClient().registry(registryParamList);
        // discovery mult consumer
        serviceRegistry.getXxlRegistryClient().discovery(registryParamKeyList);
    }

init()总结

无论是springboot还是frameless，启动client都离不开init()，init主要做三个工作
1，验证是否有效，有效的加入consumerRespository
2，初始化XxlRpcInvokerFactory，XxlRegistryServiceRegistry，ConsumerRegistryHelper和xxlMqBroker。XxlRpcInvokerFactory用于创建XxlRegistryServiceRegistry，同时开启两个线程，分别用于注册和发现数据。然后后续开启三个线程用于普通队列，三个线程用于回调队列。
3，将消费者的组，主题，uuid加入注册进去，然后将消费者主题加入发现进去，然后开启消费者线程

从xxl-mq-admin看数据库权限认证 WebMvcConfig

用于将拦截器初始化为bean，然后加入拦截器队列。

	@Configuration
	public class WebMvcConfig extends WebMvcConfigurerAdapter {
	    @Resource
	    private PermissionInterceptor permissionInterceptor;
	    @Resource
	    private CookieInterceptor cookieInterceptor;
	    @Override
	    public void addInterceptors(InterceptorRegistry registry) {
	        registry.addInterceptor(permissionInterceptor).addPathPatterns("/**");
	        registry.addInterceptor(cookieInterceptor).addPathPatterns("/**");
	        super.addInterceptors(registry);
	    }
	}

PermissionInterceptor CookieInterceptor 异常处理

如果出错了，会判断是返回JSON字符串还是视图，如果是Json直接写入response，如果不是，会返回数据并指定视图

@Override
	public ModelAndView resolveException(HttpServletRequest request,
										 HttpServletResponse response, Object handler, Exception ex) {
		logger.error("WebExceptionResolver:{}", ex);

		// if json
		boolean isJson = false;
		HandlerMethod method = (HandlerMethod)handler;
		ResponseBody responseBody = method.getMethodAnnotation(ResponseBody.class);
		if (responseBody != null) {
			isJson = true;
		}

		// error result
		ReturnT errorResult = new ReturnT(ReturnT.FAIL_CODE, ex.toString().replaceAll("n", "
"));

		// response
		ModelAndView mv = new ModelAndView();
		if (isJson) {
			try {
				response.setContentType("application/json;charset=utf-8");
				response.getWriter().print("{"code":"+errorResult.getCode()+", "msg":""+ errorResult.getMsg() +""}");
			} catch (IOException e) {
				logger.error(e.getMessage(), e);
			}
			return mv;
		} else {
			mv.addObject("exceptionMsg", errorResult.getMsg());
			mv.setViewName("/common/common.exception");
			return mv;
		}
	}

参考链接
https://www.xuxueli.com/xxl-mq/

MQ入门之看MQ项目源码 XXL-MQ

Java相关栏目本月热门文章