论坛首页 → Java企业应用论坛 →

oscache 集群和数据同步

浏览 5028 次
锁定老帖子 主题：oscache 集群和数据同步
精华帖 (0) :: 良好帖 (0) :: 新手帖 (0) :: 隐藏帖 (0)
作者 正文
  • qq123zhz
  • 等级: 二星会员
  • 性别:
  • 文章: 196
  • 积分: 220
  • 来自: 深圳
   发表时间：2011-07-29   
cache.event.listeners=com.test.JavaGroupsBroadcastingListenerImpl
cache.cluster.properties=UDP(mcast_addr=231.12.21.132;mcast_port=45566;ip_ttl=32;\
mcast_send_buf_size=150000;mcast_recv_buf_size=80000):\
PING(timeout=2000;num_initial_members=3):\
MERGE2(min_interval=5000;max_interval=10000):\
FD_SOCK:VERIFY_SUSPECT(timeout=1500):\
pbcast.NAKACK(gc_lag=50;retransmit_timeout=300,600,1200,2400,4800;max_xmit_size=8192):\
UNICAST(timeout=300,600,1200,2400):\
pbcast.STABLE(desired_avg_gossip=20000):\
FRAG(frag_size=8096;down_thread=false;up_thread=false):\
pbcast.GMS(join_timeout=5000;join_retry_timeout=2000;shun=false;print_local_addr=true)
cache.cluster.multicast.ip=231.12.21.132

 oscache 集群同步数据需要使用到JavaGroupsBroadcastingListener，但是JavaGroupsBroadcastingListener 具体没有实现同步的功能，JavaGroupsBroadcastingListener：

 public void handleNotification(Serializable serializable) {
        if (!(serializable instanceof ClusterNotification)) {
            log.error("An unknown cluster notification message received (class=" + serializable.getClass().getName() + "). Notification ignored.");

            return;
        }

        handleClusterNotification((ClusterNotification) serializable);
    }

 handleNotification该方法调用 handleClusterNotification((ClusterNotification) serializable)后，进入AbstractBoardcasting类，改方法只执行刷新当前缓存的内容，对集群数据同步没有实现。

/*
 * Copyright (c) 2002-2003 by OpenSymphony
 * All rights reserved.
 */
package com.opensymphony.oscache.plugins.clustersupport;

import com.opensymphony.oscache.base.*;
import com.opensymphony.oscache.base.events.*;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.util.Date;

/**
 * Implementation of a CacheEntryEventListener. It broadcasts the flush events
 * across a cluster to other listening caches. Note that this listener cannot
 * be used in conjection with session caches.
 *
 * @version        $Revision: 254 $
 * @author <a href="&#109;a&#105;&#108;&#116;&#111;:chris&#64;swebtec.&#99;&#111;&#109;">Chris Miller</a>
 */
public abstract class AbstractBroadcastingListener implements CacheEntryEventListener, LifecycleAware {
    private final static Log log = LogFactory.getLog(AbstractBroadcastingListener.class);

    /**
     * The name to use for the origin of cluster events. Using this ensures
     * events are not fired recursively back over the cluster.
     */
    protected static final String CLUSTER_ORIGIN = "CLUSTER";
    protected Cache cache = null;

    public AbstractBroadcastingListener() {
        if (log.isInfoEnabled()) {
            log.info("AbstractBroadcastingListener registered");
        }
    }

    /**
     * Event fired when an entry is flushed from the cache. This broadcasts
     * the flush message to any listening nodes on the network.
     */
    public void cacheEntryFlushed(CacheEntryEvent event) {
        if (!Cache.NESTED_EVENT.equals(event.getOrigin()) && !CLUSTER_ORIGIN.equals(event.getOrigin())) {
            if (log.isDebugEnabled()) {
                log.debug("cacheEntryFlushed called (" + event + ")");
            }

            sendNotification(new ClusterNotification(ClusterNotification.FLUSH_KEY, event.getKey()));
        }
    }

    /**
     * Event fired when an entry is removed from the cache. This broadcasts
     * the remove method to any listening nodes on the network, as long as
     * this event wasn't from a broadcast in the first place.
     */
    public void cacheGroupFlushed(CacheGroupEvent event) {
        if (!Cache.NESTED_EVENT.equals(event.getOrigin()) && !CLUSTER_ORIGIN.equals(event.getOrigin())) {
            if (log.isDebugEnabled()) {
                log.debug("cacheGroupFushed called (" + event + ")");
            }

            sendNotification(new ClusterNotification(ClusterNotification.FLUSH_GROUP, event.getGroup()));
        }
    }

    public void cachePatternFlushed(CachePatternEvent event) {
        if (!Cache.NESTED_EVENT.equals(event.getOrigin()) && !CLUSTER_ORIGIN.equals(event.getOrigin())) {
            if (log.isDebugEnabled()) {
                log.debug("cachePatternFushed called (" + event + ")");
            }

            sendNotification(new ClusterNotification(ClusterNotification.FLUSH_PATTERN, event.getPattern()));
        }
    }

    public void cacheFlushed(CachewideEvent event) {
        if (!Cache.NESTED_EVENT.equals(event.getOrigin()) && !CLUSTER_ORIGIN.equals(event.getOrigin())) {
            if (log.isDebugEnabled()) {
                log.debug("cacheFushed called (" + event + ")");
            }

            sendNotification(new ClusterNotification(ClusterNotification.FLUSH_CACHE, event.getDate()));
        }
    }

    // --------------------------------------------------------
    // The remaining events are of no interest to this listener
    // --------------------------------------------------------
    public void cacheEntryAdded(CacheEntryEvent event) {
    }

    public void cacheEntryRemoved(CacheEntryEvent event) {
    }

    public void cacheEntryUpdated(CacheEntryEvent event) {
    }

    public void cacheGroupAdded(CacheGroupEvent event) {
    }

    public void cacheGroupEntryAdded(CacheGroupEvent event) {
    }

    public void cacheGroupEntryRemoved(CacheGroupEvent event) {
    }

    public void cacheGroupRemoved(CacheGroupEvent event) {
    }

    public void cacheGroupUpdated(CacheGroupEvent event) {
    }

    /**
     * Called by the cache administrator class when a cache is instantiated.
     *
     * @param cache the cache instance that this listener is attached to.
     * @param config The cache's configuration details. This allows the event handler
     * to initialize itself based on the cache settings, and also to receive <em>additional</em>
     * settings that were part of the cache configuration but that the cache
     * itself does not care about. If you are using <code>cache.properties</code>
     * for your configuration, simply add any additional properties that your event
     * handler requires and they will be passed through in this parameter.
     *
     * @throws InitializationException thrown when there was a problem initializing the
     * listener. The cache administrator will log this error and disable the listener.
     */
    public void initialize(Cache cache, Config config) throws InitializationException {
        this.cache = cache;
    }

    /**
     * Handles incoming notification messages. This method should be called by the
     * underlying broadcasting implementation when a message is received from another
     * node in the cluster.
     *
     * @param message The incoming cluster notification message object.
     */
    public void handleClusterNotification(ClusterNotification message) {
        if (cache == null) {
            log.warn("A cluster notification (" + message + ") was received, but no cache is registered on this machine. Notification ignored.");

            return;
        }

        if (log.isInfoEnabled()) {
            log.info("Cluster notification (" + message + ") was received.");
        }

        switch (message.getType()) {
            case ClusterNotification.FLUSH_KEY:
                cache.flushEntry((String) message.getData(), CLUSTER_ORIGIN);
                break;
            case ClusterNotification.FLUSH_GROUP:
                cache.flushGroup((String) message.getData(), CLUSTER_ORIGIN);
                break;
            case ClusterNotification.FLUSH_PATTERN:
                cache.flushPattern((String) message.getData(), CLUSTER_ORIGIN);
                break;
            case ClusterNotification.FLUSH_CACHE:
                cache.flushAll((Date) message.getData(), CLUSTER_ORIGIN);
                break;
            default:
                log.error("The cluster notification (" + message + ") is of an unknown type. Notification ignored.");
        }
    }

    /**
     * Called when a cluster notification message is to be broadcast. Implementing
     * classes should use their underlying transport to broadcast the message across
     * the cluster.
     *
     * @param message The notification message to broadcast.
     */
    abstract protected void sendNotification(ClusterNotification message);
}

 
所以我们要集群式同步数据，必须实现这三个方法：

public void cacheEntryAdded(CacheEntryEvent event) {
    }

    public void cacheEntryRemoved(CacheEntryEvent event) {
    }

    public void cacheEntryUpdated(CacheEntryEvent event) {
    }

 这个实现类实现了以上的三个方法，就可以同步数据了：

package com.test;
import com.opensymphony.oscache.base.events.CacheEntryEvent;
import com.opensymphony.oscache.plugins.clustersupport.ClusterNotification;
import com.opensymphony.oscache.plugins.clustersupport.JavaGroupsBroadcastingListener;

public class JavaGroupsBroadcastingListenerImpl extends
		JavaGroupsBroadcastingListener {
	public void handleClusterNotification(ClusterNotification message) {
		
		switch (message.getType()) {
		case CacheConstants.CLUSTER_ENTRY_ADD:
			System.out.println("集群新增:" + message.getData());
			if(message.getData() instanceof QflagCacheEvent) {
				QflagCacheEvent event = (QflagCacheEvent)message.getData();
				cache.putInCache(event.getKey(), event.getEntry().getContent(),null,null,CLUSTER_ORIGIN);
			}
			break;
		case CacheConstants.CLUSTER_ENTRY_UPDATE:
			System.out.println("集群更新:" + message.getData());
			if(message.getData() instanceof QflagCacheEvent) {
				QflagCacheEvent event = (QflagCacheEvent)message.getData();
//				cache.flushEntry(event.getKey());
				cache.putInCache(event.getKey(), event.getEntry().getContent(),null,null,CLUSTER_ORIGIN);
			}
			break;
		case CacheConstants.CLUSTER_ENTRY_DELETE:
			System.out.println("集群删除:" + message.getData());
			if(message.getData() instanceof QflagCacheEvent) {
				QflagCacheEvent event = (QflagCacheEvent)message.getData();
//				cache.removeEntry(event.getKey(),event.getOrigin());
				cache.removeEntry(event.getKey());
			}
			break;
		}

	}
	
	public void cacheEntryAdded(CacheEntryEvent event) {
		super.cacheEntryAdded(event);
		System.out.println("属性添加");
		if(!CLUSTER_ORIGIN.equals(event.getOrigin())) {
			sendNotification(new ClusterNotification(CacheConstants.CLUSTER_ENTRY_ADD, new QflagCacheEvent(event.getMap(),event.getEntry(),CLUSTER_ORIGIN)));
		}
	}

//	@Override
//	public void cacheEntryFlushed(CacheEntryEvent event) {
//		
//		super.cacheEntryFlushed(event);
//		if(!CLUSTER_ORIGIN.equals(event.getOrigin())) {
//			sendNotification(new ClusterNotification(CacheConstants.CLUSTER_ENTRY_ADD, new UcallCacheEvent(event.getMap(),event.getEntry(),CLUSTER_ORIGIN)));
//		}
//	}

	@Override
	public void cacheEntryRemoved(CacheEntryEvent event) {
		System.out.println("属性移除");
		super.cacheEntryRemoved(event);
		if(!CLUSTER_ORIGIN.equals(event.getOrigin())) {
			sendNotification(new ClusterNotification(CacheConstants.CLUSTER_ENTRY_DELETE, new QflagCacheEvent(event.getMap(),event.getEntry(),CLUSTER_ORIGIN)));
		}
	}

	@Override
	public void cacheEntryUpdated(CacheEntryEvent event) {
		System.out.println("属性更新");
		super.cacheEntryUpdated(event);
		if(!CLUSTER_ORIGIN.equals(event.getOrigin())) {
			sendNotification(new ClusterNotification(CacheConstants.CLUSTER_ENTRY_UPDATE, new QflagCacheEvent(event.getMap(),event.getEntry(),CLUSTER_ORIGIN)));
		}
	}
	
}

 

然后修改配置文件，映射cache.event.listeners类的路径：

cache.event.listeners=com.test.JavaGroupsBroadcastingListenerImpl
cache.cluster.properties=UDP(mcast_addr=231.12.21.132;mcast_port=45566;ip_ttl=32;\
mcast_send_buf_size=150000;mcast_recv_buf_size=80000):\
PING(timeout=2000;num_initial_members=3):\
MERGE2(min_interval=5000;max_interval=10000):\
FD_SOCK:VERIFY_SUSPECT(timeout=1500):\
pbcast.NAKACK(gc_lag=50;retransmit_timeout=300,600,1200,2400,4800;max_xmit_size=8192):\
UNICAST(timeout=300,600,1200,2400):\
pbcast.STABLE(desired_avg_gossip=20000):\
FRAG(frag_size=8096;down_thread=false;up_thread=false):\
pbcast.GMS(join_timeout=5000;join_retry_timeout=2000;shun=false;print_local_addr=true)
cache.cluster.multicast.ip=231.12.21.132

 具体源码可以参考 oscache集群功能  http://rwl6813021.iteye.com/blog/246473，我已经实现了整合。

测试使用两个tomcat6.0，无问题，但是一个tomca6.0，一个tomcat7.0时发现缓存没有同步。。。
返回顶楼
         
  • diaoweili
  • 等级: 初级会员
  • 性别:
  • 文章: 13
  • 积分: 30
  • 来自: 北京
   发表时间：2011-07-29  
你都换成 tomcat6或者 都是tomcat7 就 应该没问题了
返回顶楼
          回帖地址
0 请登录后投票
  • qq123zhz
  • 等级: 二星会员
  • 性别:
  • 文章: 196
  • 积分: 220
  • 来自: 深圳
   发表时间：2011-07-30  
diaoweili 写道
你都换成 tomcat6或者 都是tomcat7 就 应该没问题了

oscache 应该和tomcat版本没关系吧。。。
返回顶楼
          回帖地址
0 请登录后投票
  • qq123zhz
  • 等级: 二星会员
  • 性别:
  • 文章: 196
  • 积分: 220
  • 来自: 深圳
   发表时间：2011-07-30  
我在想如果集群的时候，我把session都放到oscache缓存，是不是tomcat之间就不需要session复制了？。。。。
返回顶楼
          回帖地址
0 请登录后投票
  • ilrxx
  • 等级: 初级会员
  • 性别:
  • 文章: 123
  • 积分: 0
  • 来自: 北京
   发表时间：2011-08-01  
回楼上，session使用的弊端也在于此，你说的没错啦
返回顶楼
          回帖地址
0 请登录后投票
  • qq123zhz
  • 等级: 二星会员
  • 性别:
  • 文章: 196
  • 积分: 220
  • 来自: 深圳
   发表时间：2011-08-02  
ilrxx 写道
回楼上，session使用的弊端也在于此，你说的没错啦

我实现过，在集群环境下把用户登录信息存到oscache的缓存中。。。
返回顶楼
          回帖地址
0 请登录后投票
  • ieqqlin13
  • 等级: 初级会员
  • 性别:
  • 文章: 1
  • 积分: 30
  • 来自: 福州
   发表时间：2011-08-31  
qq123zhz 写道
ilrxx 写道
回楼上，session使用的弊端也在于此，你说的没错啦

我实现过，在集群环境下把用户登录信息存到oscache的缓存中。。。


那就是说，多个tomcat不用做session复制 ，也是可行啦？

返回顶楼
          回帖地址
0 请登录后投票
  • qq123zhz
  • 等级: 二星会员
  • 性别:
  • 文章: 196
  • 积分: 220
  • 来自: 深圳
   发表时间：2011-09-01  
ieqqlin13 写道
qq123zhz 写道
ilrxx 写道
回楼上，session使用的弊端也在于此，你说的没错啦

我实现过，在集群环境下把用户登录信息存到oscache的缓存中。。。


那就是说，多个tomcat不用做session复制 ，也是可行啦？


我觉得如果多个tomcat 都去缓存取值，应该是可以的。。。
返回顶楼
          回帖地址
0 请登录后投票
  • putonyuer
  • 等级: 初级会员
  • 性别:
  • 文章: 82
  • 积分: 0
  • 来自: 重庆
   发表时间：2011-09-28  
OScache 分布式的， 同步什么？ 画蛇添足？
返回顶楼
          回帖地址
0 请登录后投票
  • qq123zhz
  • 等级: 二星会员
  • 性别:
  • 文章: 196
  • 积分: 220
  • 来自: 深圳
   发表时间：2011-09-29  
putonyuer 写道
OScache 分布式的， 同步什么？ 画蛇添足？

把他当集中式的用。。。。。。。。
返回顶楼
          回帖地址
0 请登录后投票
论坛首页 → Java企业应用版

跳转论坛:
Global site tag (gtag.js) - Google Analytics