Jgroups

这是之前在网上看到的文章，跟着学习了一把。结合上一篇文章，再定义一个实体，cacheType,cacheKey,cacheValue,cacheSeconds就可以用jgroups+Guava cache实现分布式localCache的数据同步。 

package jgroups;

import org.jgroups.*;
import org.jgroups.util.Util;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Created by ZhaoYun on 2016/5/8.
 * 节点
 */
public class Node extends ReceiverAdapter {
    private static final Logger LOGGER = LoggerFactory.getLogger(Node.class);

    /**
     * 配置文件.
     */
    private static final String CONFIG_XML = "network-tcp.xml";

    /**
     * 集群名称.
     */
    private static final String CLUSTER_NAME = "ZY";

    /**
     * 节点通道.
     */
    private JChannel channel = null;

    /**
     * 以此作为节点间初始化的同步数据.
     */
    private Map<String, String> cacheData = new HashMap<String, String>();

    private ReentrantLock lock = new ReentrantLock();

    public Node(){
        InputStream is = this.getClass().getClassLoader().getResourceAsStream(CONFIG_XML);
        try {
            channel = new JChannel(is);
            channel.setReceiver(this);
            channel.connect(CLUSTER_NAME);
            channel.getState(null,50000);
        } catch (Exception e) {
            System.out.println("启动节点异常!" + e.getMessage());
            // 最好是自定义RuntimeException!
            throw new RuntimeException("启动节点异常!", e);
        }
    }

    /**
     *  发送消息给目标地址.
     * @param address 为空表示发给所有节点.
     * @param object 消息
     */
    public void sendMsg(Address address,Object object){
        Message message = new Message(address,null,object);
        try{
            channel.send(message);
        }catch (Exception e){
            System.out.println("send message error:"+e.getMessage());
            throw new RuntimeException("send message error!", e);
        }
    }

    /**
     *
     * @param output
     * @throws Exception
     */
    @Override
    public void getState(OutputStream output)throws Exception{
        lock.lock();
        try{
            Util.objectToStream(cacheData,new DataOutputStream(output));
        }catch (Exception e){
            System.out.println("get state error:"+ e.getMessage());
            throw new RuntimeException();
        }finally {
            lock.unlock();
        }
    }

    /**
     *
     * @param message
     */
    @Override
    public void receive(Message message){
        //当前节点不接收自己发送到通道当中的消息.
        if(message.getSrc().equals(channel.getAddress())){
            System.out.println(" self ");
            return;
        }
        System.out.println(message.getObject()+ "ZY"+message.getDest());
    }

    /**
     *
     * @param inputStream
     * @throws Exception
     */
    @Override
    public void setState(InputStream inputStream) throws Exception{
        lock.lock();
        try{
            Map<String, String> cacheData = (Map<String, String>) Util.objectFromStream(new DataInputStream(inputStream));
            this.cacheData.putAll(cacheData);
        }catch (Exception e){
            System.out.println("从主节点同步状态到当前节点发生异常!" + e.getMessage());
        }finally {
            lock.unlock();
        }
    }

    @Override
    public void viewAccepted(View view) {
        System.out.println("当前成员[" + this.channel.getAddressAsString() + "]");
        System.out.println(view.getCreator());
        System.out.println(view.getMembers());
        System.out.println("当前节点数据:" + cacheData);
    }

    /**
     * 提供一个简单的初始化数据的方法.
     * @param key
     * @param val
     */
    public void addData(String key,String val){
        if(key!=null&&!key.isEmpty()){
            cacheData.put(key, val);
        }
    }

Jgroups maven依赖：

<!-- jgroups start  用于多实例之间信息同步-->
        <dependency>
            <groupId>org.jgroups</groupId>
            <artifactId>jgroups</artifactId>
            <version>3.5.0.Final</version>
        </dependency>
        <!-- jgroups end -->

 

转载于:https://my.oschina.net/u/1438609/blog/673594