Coherence代理节点在离开集群时的恢复

Posted 2020-08-11 ericnie的技术博客

Coherence的架构参考

 

在极端压力之下，有时候代理节点会忙于处理请求而不响应其他的心跳，同步，导致其他节点传输的报文没有回应，而被认为是离开集群，从而影响业务。

写了一段代码，能让进程在监听到有节点离开时关闭节点，同时通过命令自动重起，实现恢复功能。

 

其中有几个要点问题解决如下:

1.Coherence Server可能是多台机器，这样任何proxy离开都会发送消息到监听程序，监听程序需要判断是否是本地进程才能操作。

2.如果通过ip来判断，java在获取本地ip时更多时候是一个list列表，所以程序中通过hostname进行判断

3.hostname和ip的映射，因为只涉及2台coherence server,所以直接把这种关系固定在程序中，当然也可以放到数据库，coherence或者文件。

4.通过proxy离开的消息能够获取processid,但通过pid如何可以获取proxy监听的端口，这里是通过

netstat -nap |grep "+processid +" | grep tcp | grep "+ ip+ ":9"的操作，也就是约定监听在以9开头的端口，然后把特定的行取出来后再进行解析。

5.获取port后再根据port和程序的映射关系去运行相关的启动命令。

6.测试可以分开测试，比如先是获取进程号的测试，然后再进行有进程号后如何kill和重新启动的测试. main下面好多都是调试的脚本。

代码如下：

 

package coherencetest; import com.tangosol.net.CacheFactory; import com.tangosol.net.MemberEvent; import com.tangosol.net.MemberListener; import com.tangosol.net.NamedCache; import com.tangosol.util.MapEvent; import com.tangosol.util.MapListener; import java.io.InputStreamReader; import java.io.LineNumberReader; import java.net.InetAddress; import java.util.ArrayList; import java.util.List; class ProxyListenerNotification implements MemberListener { static String IP1 = "192.168.0.150"; String ip; public void memberJoined (MemberEvent e) { System.out.println("====== Member Join "); } public void memberLeaving(MemberEvent e) {} public void memberLeft(MemberEvent e) { System.out.println("************ Member Left"); System.out.println("===================="); // System.out.println(e.toString()); System.out.println(e.getMember()); String processId = e.getMember().getProcessName(); String totalstring = e.getMember().getAddress().getHostName(); String localhostname =""; try { InetAddress addr = InetAddress.getLocalHost(); localhostname=addr.getHostName().toString(); System.out.println("=============*======="+ip); } catch (Exception e1) { System.out.println(e1.getMessage()); } if (totalstring.indexOf(localhostname) != 0) { if (localhostname == "ocp") { ip=IP1; } String shStr; String port = getPort(Integer.parseInt(processId),ip); if ( port =="9099") { shStr = "/home/oracle/Middleware/coherence/bin/proxy-cache.sh"; } else { shStr = "/home/oracle/Middleware/coherence/bin/proxy-cache1.sh"; } Process process; String killStr = "kill -9 "+processId; try { process = Runtime.getRuntime().exec(killStr); System.out.println("kill successful"); process = Runtime.getRuntime().exec(new String[]{"/bin/sh","-c",shStr},null,null); } catch(Exception e1) { System.out.println(e1.getMessage()); } } } public String getPort(int processid,String ip) { Process process; String line=""; try { String shStr = "netstat -nap |grep "+processid +" | grep tcp | grep "+ ip+ ":9"; System.out.println(shStr); process = Runtime.getRuntime().exec(new String[]{"/bin/sh","-c",shStr},null,null); InputStreamReader ir = new InputStreamReader(process.getInputStream()); LineNumberReader input = new LineNumberReader(ir); process.waitFor(); line = input.readLine(); /*while ((line = input.readLine()) != null){ strList.add(line); }*/ }catch (Exception e) { System.out.println(e.getMessage()); } int index=line.indexOf(":9"); String port = line.substring(index+1, index+5); System.out.println(" port ="+port); return port; } }   public class ProxyListener { public static void main(String[] args) { try { NamedCache cache = CacheFactory.getCache("POFSample"); /* cache.addMapListener(new MapListener() { public void entryUpdated(MapEvent arg0) { System.out.println(arg0); } public void entryInserted(MapEvent arg0) { System.out.println(arg0); } public void entryDeleted(MapEvent arg0) { System.out.println(arg0); }}); */ String processId = "6458"; String localhostname =""; try { InetAddress addr = InetAddress.getLocalHost(); localhostname=addr.getHostName().toString(); System.out.println("=============*======="+localhostname); } catch (Exception e1) { System.out.println(e1.getMessage()); } String ip = "192.168.0.150"; String shStr; ProxyListenerNotification pl = new ProxyListenerNotification(); String port = pl.getPort(Integer.parseInt(processId),ip); //cache.getCacheService().addMemberListener(new ProxyListenerNotification()); if ( port.equals("9099")) { System.out.println("1"); shStr = "/home/oracle/Middleware/coherence/bin/proxy-cache.sh"; } else { System.out.println("2"); shStr = "/home/oracle/Middleware/coherence/bin/proxy-cache1.sh"; } Process process; String killStr = "kill -9 "+processId; try { process = Runtime.getRuntime().exec(killStr); System.out.println("kill successful"); process = Runtime.getRuntime().exec(new String[]{"/bin/sh","-c",shStr},null,null); } catch(Exception e1) { System.out.println(e1.getMessage()); } while(true){ // Thread.sleep(100000); } } catch (Exception e) { System.out.println(e); } } }