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);
}
}
}
|