Java web servers 间是如何实现 session 同步的
有一个多月的时间没有更新博客了，今天终于忙里偷闲，可以把近期的收获总结一下。
本文是关于Java web servers 之间是如何实现 session 同步的，其实其他技术栈也面临同样的问题需要解决，而且大部分场景下已经有了成熟的解决方案，其实就应用开发本身，大部分人不太会关注这个问题，因为我们大部分人写代码的时候只需要考虑单节点场景，其他同步部分由服务器端负责实现，但是作为一个刨根问底的人，可能这个问题本身已经能够吸引人的了。
那么，为了解决这个问题，有哪些关键点呢，下面的几点可能是我们绕不开的，
1. 如何保证可靠传输呢，也就是说发送端确认接收节点收到了session数据
2. 一个节点如何知道他自己有哪些伙伴，他需要把session数据发给谁呢
3. 长消息如何发送呢，如何保证数据安全传输
写到这里，大家可能脑海中已经出现了可靠传输，IP多播，数据分包，加密解密，数据一致性保证，对的，就是这些技术，但是应用这些底层技术完成应用，确实需要不是一般程序员可以负担起的时间和经历。笔者也不打算展开来讲所有的技术细节，经过简单的研究，笔者发现了一个写的比较好的开源框架，可以完成所有相关的功能，下面就基于这个开源框架谈谈session同步是如何做到的。示例代码和效果如下，当我在第一张面板上写下tea的时候，在其他所用同一个组的面板上也会显示出同样的字样，同样的效果，JBoss cluster 和JBoss Cache都是基于此开源框架进行的实现，此开源框架的名字是 JGroups 。
1 public class Draw extends ReceiverAdapter implements ActionListener, ChannelListener { 2 protected String cluster_name="draw"; 3 private JChannel channel=null; 4 private int member_size=1; 5 private JFrame mainFrame=null; 6 private JPanel sub_panel=null; 7 private DrawPanel panel=null; 8 private JButton clear_button, leave_button; 9 private final Random random=new Random(System.currentTimeMillis()); 10 private final Font default_font=new Font("Helvetica",Font.PLAIN,12); 11 private final Color draw_color=selectColor(); 12 private static final Color background_color=Color.white; 13 boolean no_channel=false; 14 boolean jmx; 15 private boolean use_state=false; 16 private long state_timeout=5000; 17 private boolean use_unicasts=false; 18 protected boolean send_own_state_on_merge=true; 19 private final List members=new ArrayList<>(); 20 21 22 public Draw(String props, boolean no_channel, boolean jmx, boolean use_state, long state_timeout, 23 boolean use_unicasts, String name, boolean send_own_state_on_merge, AddressGenerator gen) throws Exception { 24 this.no_channel=no_channel; 25 this.jmx=jmx; 26 this.use_state=use_state; 27 this.state_timeout=state_timeout; 28 this.use_unicasts=use_unicasts; 29 if(no_channel) 30 return; 31 32 channel=new JChannel(props).addAddressGenerator(gen).setName(name); 33 channel.setReceiver(this).addChannelListener(this); 34 this.send_own_state_on_merge=send_own_state_on_merge; 35 } 36 37 public Draw(JChannel channel) throws Exception { 38 this.channel=channel; 39 channel.setReceiver(this); 40 channel.addChannelListener(this); 41 } 42 43 44 public Draw(JChannel channel, boolean use_state, long state_timeout) throws Exception { 45 this.channel=channel; 46 channel.setReceiver(this); 47 channel.addChannelListener(this); 48 this.use_state=use_state; 49 this.state_timeout=state_timeout; 50 } 51 52 53 public String getClusterName() { 54 return cluster_name; 55 } 56 57 public void setClusterName(String clustername) { 58 if(clustername != null) 59 this.cluster_name=clustername; 60 } 61 62 63 public static void main(String[] args) { 64 Draw draw=null; 65 String props=null; 66 boolean no_channel=false; 67 boolean jmx=true; 68 boolean use_state=false; 69 String group_name=null; 70 long state_timeout=5000; 71 boolean use_unicasts=false; 72 String name=null; 73 boolean send_own_state_on_merge=true; 74 AddressGenerator generator=null; 75 76 for(int i=0; i < args.length; i++) { 77 if("-help".equals(args[i])) { 78 help(); 79 return; 80 } 81 if("-props".equals(args[i])) { 82 props=args[++i]; 83 continue; 84 } 85 if("-no_channel".equals(args[i])) { 86 no_channel=true; 87 continue; 88 } 89 if("-jmx".equals(args[i])) { 90 jmx=Boolean.parseBoolean(args[++i]); 91 continue; 92 } 93 if("-clustername".equals(args[i])) { 94 group_name=args[++i]; 95 continue; 96 } 97 if("-state".equals(args[i])) { 98 use_state=true; 99 continue;100 }101 if("-timeout".equals(args[i])) {102 state_timeout=Long.parseLong(args[++i]);103 continue;104 }105 if("-bind_addr".equals(args[i])) {106 System.setProperty("jgroups.bind_addr", args[++i]);107 continue;108 }109 if("-use_unicasts".equals(args[i])) {110 use_unicasts=true;111 continue;112 }113 if("-name".equals(args[i])) {114 name=args[++i];115 continue;116 }117 if("-send_own_state_on_merge".equals(args[i])) {118 send_own_state_on_merge=Boolean.getBoolean(args[++i]);119 continue;120 }121 if("-uuid".equals(args[i])) {122 generator=new OneTimeAddressGenerator(Long.valueOf(args[++i]));123 continue;124 }125 126 help();127 return;128 }129 130 try {131 draw=new Draw(props, no_channel, jmx, use_state, state_timeout, use_unicasts, name,132 send_own_state_on_merge, generator);133 if(group_name != null)134 draw.setClusterName(group_name);135 draw.go();136 }137 catch(Throwable e) {138 e.printStackTrace(System.err);139 System.exit(0);140 }141 }142 143 144 static void help() {145 System.out.println("\nDraw [-help] [-no_channel] [-props ]" +146 " [-clustername ] [-state] [-timeout ] [-use_unicasts] " +147 "[-bind_addr ] [-jmx ] [-name ] [-send_own_state_on_merge true|false] " +148 "[-uuid ]");149 System.out.println("-no_channel: doesn't use JGroups at all, any drawing will be relected on the " +150 "whiteboard directly");151 System.out.println("-props: argument can be an old-style protocol stack specification, or it can be " +152 "a URL. In the latter case, the protocol specification will be read from the URL\n");153 }154 155 156 private Color selectColor() {157 int red=Math.abs(random.nextInt() % 255);158 int green=Math.abs(random.nextInt() % 255);159 int blue=Math.abs(random.nextInt() % 255);160 return new Color(red, green, blue);161 }162 163 164 private void sendToAll(byte[] buf) throws Exception {165 for(Address mbr: members)166 channel.send(new Message(mbr, buf));167 }168 169 170 public void go() throws Exception {171 if(!no_channel && !use_state)172 channel.connect(cluster_name);173 mainFrame=new JFrame();174 mainFrame.setDefaultCloseOperation(WindowConstants.EXIT_ON_CLOSE);175 panel=new DrawPanel(use_state);176 panel.setBackground(background_color);177 sub_panel=new JPanel();178 mainFrame.getContentPane().add("Center", panel);179 clear_button=new JButton("Clear");180 clear_button.setFont(default_font);181 clear_button.addActionListener(this);182 leave_button=new JButton("Leave");183 leave_button.setFont(default_font);184 leave_button.addActionListener(this);185 sub_panel.add("South", clear_button);186 sub_panel.add("South", leave_button);187 mainFrame.getContentPane().add("South", sub_panel);188 mainFrame.setBackground(background_color);189 clear_button.setForeground(Color.blue);190 leave_button.setForeground(Color.blue);191 mainFrame.pack();192 mainFrame.setLocation(15, 25);193 mainFrame.setBounds(new Rectangle(250, 250));194 195 if(!no_channel && use_state) {196 channel.connect(cluster_name, null, state_timeout);197 }198 mainFrame.setVisible(true);199 setTitle();200 }201 202 203 204 205 void setTitle(String title) {206 String tmp="";207 if(no_channel) {208 mainFrame.setTitle(" Draw Demo ");209 return;210 }211 if(title != null) {212 mainFrame.setTitle(title);213 }214 else {215 if(channel.getAddress() != null)216 tmp+=channel.getAddress();217 tmp+=" (" + member_size + ")";218 mainFrame.setTitle(tmp);219 }220 }221 222 void setTitle() {223 setTitle(null);224 }225 226 public void receive(Message msg) {227 byte[] buf=msg.getRawBuffer();228 if(buf == null) {229 System.err.printf("%s: received null buffer from %s, headers: %s\n", channel.getAddress(), msg.src(), msg.printHeaders());230 return;231 }232 233 try {234 DrawCommand comm=Util.streamableFromByteBuffer(DrawCommand.class, buf, msg.getOffset(), msg.getLength());235 switch(comm.mode) {236 case DrawCommand.DRAW:237 if(panel != null)238 panel.drawPoint(comm);239 break;240 case DrawCommand.CLEAR:241 clearPanel();242 break;243 default:244 System.err.println("***** received invalid draw command " + comm.mode);245 break;246 }247 }248 catch(Exception e) {249 e.printStackTrace();250 }251 }252 253 public void viewAccepted(View v) {254 member_size=v.size();255 if(mainFrame != null)256 setTitle();257 members.clear();258 members.addAll(v.getMembers());259 260 if(v instanceof MergeView) {261 System.out.println("** " + v);262 263 // This is an example of a simple merge function, which fetches the state from the coordinator264 // on a merge and overwrites all of its own state265 if(use_state && !members.isEmpty()) {266 Address coord=members.get(0);267 Address local_addr=channel.getAddress();268 if(local_addr != null && !local_addr.equals(coord)) {269 try {270 271 // make a copy of our state first272 Map copy=null;273 if(send_own_state_on_merge) {274 synchronized(panel.state) {275 copy=new LinkedHashMap<>(panel.state);276 }277 }278 System.out.println("fetching state from " + coord);279 channel.getState(coord, 5000);280 if(copy != null)281 sendOwnState(copy); // multicast my own state so everybody else has it too282 }283 catch(Exception e) {284 e.printStackTrace();285 }286 }287 }288 }289 else290 System.out.println("** View=" + v);291 }292 293 294 public void getState(OutputStream ostream) throws Exception {295 panel.writeState(ostream);296 }297 298 public void setState(InputStream istream) throws Exception {299 panel.readState(istream);300 }301 302 /* --------------- Callbacks --------------- */303 304 305 306 public void clearPanel() {307 if(panel != null)308 panel.clear();309 }310 311 public void sendClearPanelMsg() {312 DrawCommand comm=new DrawCommand(DrawCommand.CLEAR);313 try {314 byte[] buf=Util.streamableToByteBuffer(comm);315 if(use_unicasts)316 sendToAll(buf);317 else318 channel.send(new Message(null, buf));319 }320 catch(Exception ex) {321 System.err.println(ex);322 }323 }324 325 326 public void actionPerformed(ActionEvent e) {327 String command=e.getActionCommand();328 switch(command) {329 case "Clear":330 if(no_channel) {331 clearPanel();332 return;333 }334 sendClearPanelMsg();335 break;336 case "Leave":337 stop();338 break;339 default:340 System.out.println("Unknown action");341 break;342 }343 }344 345 346 public void stop() {347 if(!no_channel) {348 try {349 channel.close();350 }351 catch(Exception ex) {352 System.err.println(ex);353 }354 }355 mainFrame.setVisible(false);356 mainFrame.dispose();357 }358 359 protected void sendOwnState(final Map copy) {360 if(copy == null)361 return;362 for(Point point: copy.keySet()) {363 // we don't need the color: it is our draw_color anyway364 DrawCommand comm=new DrawCommand(DrawCommand.DRAW, point.x, point.y, draw_color.getRGB());365 try {366 byte[] buf=Util.streamableToByteBuffer(comm);367 if(use_unicasts)368 sendToAll(buf);369 else370 channel.send(new Message(null, buf));371 }372 catch(Exception ex) {373 System.err.println(ex);374 }375 }376 }377 378 379 /* ------------------------------ ChannelListener interface -------------------------- */380 381 public void channelConnected(JChannel channel) {382 if(jmx) {383 Util.registerChannel(channel, "jgroups");384 }385 }386 387 public void channelDisconnected(JChannel channel) {388 if(jmx) {389 MBeanServer server=Util.getMBeanServer();390 if(server != null) {391 try {392 JmxConfigurator.unregisterChannel(channel, server, cluster_name);393 }394 catch(Exception e) {395 e.printStackTrace();396 }397 }398 }399 }400 401 public void channelClosed(JChannel channel) {402 403 }404 405 406 /* --------------------------- End of ChannelListener interface ---------------------- */407 408 409 410 protected class DrawPanel extends JPanel implements MouseMotionListener {411 protected final Dimension preferred_size=new Dimension(235, 170);412 protected Image img; // for drawing pixels413 protected Dimension d, imgsize;414 protected Graphics gr;415 protected final Map state;416 417 418 public DrawPanel(boolean use_state) {419 if(use_state)420 state=new LinkedHashMap<>();421 else422 state=null;423 createOffscreenImage(false);424 addMouseMotionListener(this);425 addComponentListener(new ComponentAdapter() {426 public void componentResized(ComponentEvent e) {427 if(getWidth() <= 0 || getHeight() <= 0) return;428 createOffscreenImage(false);429 }430 });431 }432 433 434 public void writeState(OutputStream outstream) throws IOException {435 if(state == null)436 return;437 synchronized(state) {438 DataOutputStream dos=new DataOutputStream(new BufferedOutputStream(outstream));439 // DataOutputStream dos=new DataOutputStream(outstream);440 dos.writeInt(state.size());441 for(Map.Entry entry: state.entrySet()) {442 Point point=entry.getKey();443 Color col=entry.getValue();444 dos.writeInt(point.x);445 dos.writeInt(point.y);446 dos.writeInt(col.getRGB());447 }448 dos.flush();449 System.out.println("wrote " + state.size() + " elements");450 }451 }452 453 454 public void readState(InputStream instream) throws IOException {455 DataInputStream in=new DataInputStream(new BufferedInputStream(instream));456 Map new_state=new LinkedHashMap<>();457 int num=in.readInt();458 for(int i=0; i < num; i++) {459 Point point=new Point(in.readInt(), in.readInt());460 Color col=new Color(in.readInt());461 new_state.put(point, col);462 }463 464 synchronized(state) {465 state.clear();466 state.putAll(new_state);467 System.out.println("read " + state.size() + " elements");468 createOffscreenImage(true);469 }470 }471 472 473 void createOffscreenImage(boolean discard_image) {474 d=getSize();475 if(discard_image) {476 img=null;477 imgsize=null;478 }479 if(img == null || imgsize == null || imgsize.width != d.width || imgsize.height != d.height) {480 img=createImage(d.width, d.height);481 if(img != null) {482 gr=img.getGraphics();483 if(gr != null && state != null) {484 drawState();485 }486 }487 imgsize=d;488 }489 repaint();490 }491 492 493 /* ---------------------- MouseMotionListener interface------------------------- */494 495 public void mouseMoved(MouseEvent e) {}496 497 public void mouseDragged(MouseEvent e) {498 int x=e.getX(), y=e.getY();499 DrawCommand comm=new DrawCommand(DrawCommand.DRAW, x, y, draw_color.getRGB());500 501 if(no_channel) {502 drawPoint(comm);503 return;504 }505 506 try {507 byte[] buf=Util.streamableToByteBuffer(comm);508 if(use_unicasts)509 sendToAll(buf);510 else511 channel.send(new Message(null, buf));512 }513 catch(Exception ex) {514 System.err.println(ex);515 }516 }517 518 /* ------------------- End of MouseMotionListener interface --------------------- */519 520 521 /**522 * Adds pixel to queue and calls repaint() whenever we have MAX_ITEMS pixels in the queue523 * or when MAX_TIME msecs have elapsed (whichever comes first). The advantage compared to just calling524 * repaint() after adding a pixel to the queue is that repaint() can most often draw multiple points525 * at the same time.526 */527 public void drawPoint(DrawCommand c) {528 if(c == null || gr == null) return;529 Color col=new Color(c.rgb);530 gr.setColor(col);531 gr.fillOval(c.x, c.y, 10, 10);532 repaint();533 if(state != null) {534 synchronized(state) {535 state.put(new Point(c.x, c.y), col);536 }537 }538 }539 540 541 542 public void clear() {543 if(gr == null) return;544 gr.clearRect(0, 0, getSize().width, getSize().height);545 repaint();546 if(state != null) {547 synchronized(state) {548 state.clear();549 }550 }551 }552 553 554 /** Draw the entire panel from the state */555 public void drawState() {556 // clear();557 Map.Entry entry;558 Point pt;559 Color col;560 synchronized(state) {561 for(Iterator it=state.entrySet().iterator(); it.hasNext();) {562 entry=(Map.Entry)it.next();563 pt=(Point)entry.getKey();564 col=(Color)entry.getValue();565 gr.setColor(col);566 gr.fillOval(pt.x, pt.y, 10, 10);567 568 }569 }570 repaint();571 }572 573 574 public Dimension getPreferredSize() {575 return preferred_size;576 }577 578 579 public void paintComponent(Graphics g) {580 super.paintComponent(g);581 if(img != null) {582 g.drawImage(img, 0, 0, null);583 }584 }585 586 }587 588 }
我们甚至可以通过如下短短的几行代码写一个简易的聊天程序，这样，一个人发送的消息，组内所有成员都可以收到，并且可以同步聊天记录，同时组内节点可以感知道其他节点的加入，关闭，甚至意外退出。
1 public class SimpleChat extends ReceiverAdapter { 2 JChannel channel; 3 String user_name=System.getProperty("user.name", "n/a"); 4 final List state=new LinkedList<>(); 5 6 public void viewAccepted(View new_view) { 7 System.out.println("** view: " + new_view); 8 } 9 10 public void receive(Message msg) {11 String line=msg.getSrc() + ": " + msg.getObject();12 System.out.println(line);13 synchronized(state) {14 state.add(line);15 }16 }17 18 public void getState(OutputStream output) throws Exception {19 synchronized(state) {20 Util.objectToStream(state, new DataOutputStream(output));21 }22 }23 24 @SuppressWarnings("unchecked")25 public void setState(InputStream input) throws Exception {26 List list=Util.objectFromStream(new DataInputStream(input));27 synchronized(state) {28 state.clear();29 state.addAll(list);30 }31 System.out.println("received state (" + list.size() + " messages in chat history):");32 list.forEach(System.out::println);33 }34 35 36 private void start() throws Exception {37 channel=new JChannel().setReceiver(this);38 channel.connect("ChatCluster");39 channel.getState(null, 10000);40 eventLoop();41 channel.close();42 }43 44 private void eventLoop() {45 BufferedReader in=new BufferedReader(new InputStreamReader(System.in));46 while(true) {47 try {48 System.out.print("> "); System.out.flush();49 String line=in.readLine().toLowerCase();50 if(line.startsWith("quit") || line.startsWith("exit")) {51 break;52 }53 line="[" + user_name + "] " + line;54 Message msg=new Message(null, line);55 channel.send(msg);56 }57 catch(Exception e) {58 }59 }60 }61 62 63 public static void main(String[] args) throws Exception {64 new SimpleChat().start();65 }66 }
总结
本文通过两个简单的示例展示了JGroups的用法，说明了 Java web servers 间是实现 session 同步的基本原理，大家如果对更多的细节感兴趣，可以和笔者进行沟通，笔者可以在下次的文章中加入更多的细节。