HornetQ集群广播解析
HornetQ支持集群方式来支持扩展性，集群中的节点借助JGroup组件来进行节点间的通信。在UTP方式中，节点在间隔时间内不断的像广播地址中发送消息，而客户端的程序就可以通过监听这个广播地址，解析广播的内容就能够知道集群中有哪些主机。对于使用Java语言的开发者来说，HornetQ的JAVA客户端相关类屏蔽了这些底层代码，使得开发变得简单。然而对于其他语言，需要自己手动解析相关内容。
以Hornet2.1.2版本为准，HornetQ每个节点像广播地址广播的信息遵循以下格式：
节点名称的长度+节点名称+唯一标识长度+唯一标识+连接器个数+各个连接器信息
下面是解析的JAVA代码，包含的内部类主要负责控制读取。
客户端的调用只需要启动该类，然后获取hostInfoMap就能够获得集群中的主机和他们的连接信息以及备用连接信息。如果要客户端的主机信息能够实时更新，可以通过修改主机集合向客户端发消息的方式实现
/* * Discovery .java * 本类非线程安全 */package com.socket.multisocket;import java.io.IOException;import java.net.DatagramPacket;import java.net.InetAddress;import java.net.MulticastSocket;import java.net.UnknownHostException;import java.util.ArrayList;import java.util.HashMap;import java.util.HashSet;import java.util.List;import java.util.Map;import java.util.Set;import java.util.Map.Entry;public class Discovery extends Thread { private static final String HOST = "host"; private static final String PORT = "port"; private static final long timeout = Integer.valueOf(System.getProperty("broadcast_host_timeout", "10000")); private static final String spliter = ":"; //存储节点信息 private static Map<String,List<Map<String,String>>> hostInfoMap = new HashMap<String,List<Map<String,String>>>(); //存储节点名称，主要是定时刷新 public final static Map<String, Long> connectors = new HashMap<String, Long>(); /** * 当前读的位置 */ private static int currentIndex = 0; public static int connectorPairsSize = 0; private String broadcastAddress; private int broadcastPort; public Discovery(String broadcastAddress, int broadcastPort) { super(); this.broadcastAddress = broadcastAddress; this.broadcastPort = broadcastPort; } public static Map<String,List<Map<String,String>>> getHost(){ return hostInfoMap; } /** * 重置当前读指针 ** */ public void putConnector(String key, Long currentTime) { connectors.put(key, currentTime); } public void validateConnectors() { Set<Entry<String, Long>> set = new HashSet<Entry<String,Long>>(connectors.entrySet()); for (Entry<String, Long> entry : set) { if (entry.getValue() + timeout < System.currentTimeMillis()) { connectors.remove(entry.getKey()); hostInfoMap.remove(entry.getKey()); } } } public void run() { InetAddress group = null; MulticastSocket server = null; try { group = InetAddress.getByName(broadcastAddress); server = new MulticastSocket(broadcastPort); server.joinGroup(group); final byte[] data = new byte[65535]; DatagramPacket recv = new DatagramPacket(data, data.length); for (;;) { server.receive(recv); String uniqueName = ParseRecieveData.parsebytes(recv .getData()); ParseRecieveData.release(); putConnector(uniqueName, System .currentTimeMillis()); validateConnectors(); } } catch (UnknownHostException e) { throw new RuntimeException("can not find this broadcast address:"+broadcastAddress); } catch (IOException e) { e.printStackTrace(); } finally { if (server != null) try { server.leaveGroup(group); } catch (IOException e) { } } } private static class ParseRecieveData { private static final byte TYPE_BOOLEAN = 0; private static final byte TYPE_INT = 1; private static final byte TYPE_LONG = 2; private static final byte TYPE_STRING = 3; /** * 解析报文 * * @author Jombo * @param bytes */ public static String parsebytes(byte[] bytes) { int nodeIdLength = getInt(bytes, currentIndex); getString(nodeIdLength, bytes, currentIndex); int uniqueIdLength = getInt(bytes, currentIndex); String uniqueName = getString(uniqueIdLength, bytes, currentIndex); connectorPairsSize = getInt(bytes, currentIndex); List<Map<String,String>> connectorList = new ArrayList<Map<String,String>>(); // connectorPairsSize可能有多个 for(int i =0 ;i<connectorPairsSize;i++){ Map<String, String> connectorPair = parseConnectorPairInfo(bytes); connectorList.add(connectorPair); } hostInfoMap.put(uniqueName, connectorList); return uniqueName; } /** * 解析连接器信息 * * @author Jombo * @param bytes */ private static Map<String, String> parseConnectorPairInfo(byte[] bytes) { Map<String, String> ConnectorPairMap = parseConnectorInfo(bytes); Map<String, String> mainConnectorMap = parseConnectorInfo(bytes); byte existBackup = getByte(bytes, currentIndex); boolean existbackupConnector = existBackup == 0 ? false : true; Map<String, String> backConnectorMap = null; if(existbackupConnector){ backConnectorMap = parseConnectorInfo(bytes); } //在这里我们只想看host+port的形式，有兴趣的朋友可以把全部信息放进去 //这是为了获取主备连接 String mainconnector = mainConnectorMap.get(HOST) + spliter + mainConnectorMap.get(PORT); String backconnector = backConnectorMap ==null ? null:backConnectorMap.get(HOST) + spliter + backConnectorMap.get(PORT); ConnectorPairMap.put(mainconnector, backconnector); return ConnectorPairMap; }private static Map<String, String> parseConnectorInfo(byte[] bytes) {Map<String, String> paramMap = new HashMap<String, String>(); int namelength = getInt(bytes, currentIndex); getString(namelength, bytes, currentIndex); int factorylength = getInt(bytes, currentIndex); getString(factorylength, bytes, currentIndex); int paramPairsSize = getInt(bytes, currentIndex); for (int j = 0; j < paramPairsSize; j++) { int keylength = getInt(bytes, currentIndex); String key = getString(keylength, bytes, currentIndex); byte valuetype = getByte(bytes, currentIndex); String value = ""; if (valuetype == TYPE_STRING) { int valuelength = getInt(bytes, currentIndex); value = getString(valuelength, bytes, currentIndex); } else if (valuetype == TYPE_BOOLEAN) { byte booleanvalue = getByte(bytes, currentIndex); value = booleanvalue == 0 ? "false" : "true"; } else if (valuetype == TYPE_INT) { value = String.valueOf(getInt(bytes, currentIndex)); } else if (valuetype == TYPE_LONG) { value = String.valueOf(getLong(bytes, currentIndex)); } else { throw new RuntimeException("invalid type"); } paramMap.put(key, value); }return paramMap;} /** * 读取一个Int * * @author Jombo * @param bytes * @param index * begin index * @return */ public static int getInt(byte[] bytes, int index) { int val = (bytes[index] & 0xff) << 24 | (bytes[index + 1] & 0xff) << 16 | (bytes[index + 2] & 0xff) << 8 | (bytes[index + 3] & 0xff) << 0; setCurrentIndex(index + 4); return val; } /** * 读取一个Byte * * @author Jombo * @param bytes * @param index * @return */ public static byte getByte(byte[] bytes, int index) { byte b = bytes[index]; setCurrentIndex(index + 1); return b; } /** * 读取Short * * @author Jombo * @param bytes * @param index * @return */ public static short getShort(byte[] bytes, int index) { short indexshort = (short) (bytes[index] << 8 | bytes[index + 1] & 0xFF); setCurrentIndex(index + 2); return indexshort; } /** * 读取字符串，根据字符串的长度采取不同的格式 * * @author Jombo * @param length * @param bytes * @param index * @return */ public static String getString(int length, byte[] bytes, int index) { StringBuffer stringBuffer = new StringBuffer(); if (length < 9) { for (int i = 0; i < length; i++) { short indexshort = getShort(bytes, index); stringBuffer.append((char) indexshort); index = index + 2; } } else if (length < 0xfff) { short utflen = getShort(bytes, index); index = index + 2; byte[] strbyte = new byte[utflen]; System.arraycopy(bytes, index, strbyte, 0, utflen); stringBuffer.append(new String(strbyte)); setCurrentIndex(index + utflen); } else { int longlength = getInt(bytes, index); index = index + 4; byte[] strbyte = new byte[longlength]; System.arraycopy(bytes, index, strbyte, 0, longlength); stringBuffer.append(new String(strbyte)); setCurrentIndex(index + longlength); } return stringBuffer.toString(); } /** * 读取Long * * @author Jombo * @param bytes * @param index * @return */ public static long getLong(byte[] bytes, int index) { setCurrentIndex(index + 8); return ((long) bytes[index] & 0xff) << 56 | ((long) bytes[index + 1] & 0xff) << 48 | ((long) bytes[index + 2] & 0xff) << 40 | ((long) bytes[index + 3] & 0xff) << 32 | ((long) bytes[index + 4] & 0xff) << 24 | ((long) bytes[index + 5] & 0xff) << 16 | ((long) bytes[index + 6] & 0xff) << 8 | ((long) bytes[index + 7] & 0xff) << 0; } /** * 重置当前读指针 ** */ public static void release() { currentIndex = 0; } public static void setCurrentIndex(int index) { currentIndex = index; } }}