野外如何利用jgroups实现分布式环境下消息的接受和发送
时间:2008-02-23 09:36来源:互联网
为了提高应用的性能,我们准备实现分布式cache,所以我特别研究了oscache自制花盆关于分布式实现的部分. 我们知道为了实现分布式环境下消息的通知,目前两种比较流行的做法是使用JavaGroups[]和JMS。这两种方式都在底层实现了广播发布消息。 由于JGroups可
为了提高应用的性能,我们准备实现分布式cache,所以我特别研究了拉布拉多小狗oscache关于分布式实现的部分.
我们知道为了实现分布式环境下消息的通知,目前两种比较流行的做法是使用JavaGroups[]和JMS。这两种方式都在底层实现了广播发布消息。
由于JGroups可以提供可靠的广播通信环绕近义词.所以我们准备采用JGroups.
我自己写了一个JavaGrouPBroadcastingManager.java类实现消息的管理(包括发送和接收),
代码参考了oscache的相关代码,在其基础上进行了改进.
代码如下:
1、JavaGroupBroadcastingManager.java
;
import com.opensymphony.oscache.ba.FinalizationException;
import com.opensymphony.oscache.ba.InitializationException;
import com.opensymphony.oscache.plugins.clustersupport.JavaGroupsBroadcastingListener;
import s.logging.Log;
import s.logging.LogFactory;
import org.jgroups.Address;
import org.jgroups.Channel;
import org.jgroups.blocks.NotificationBus;
import java.io.Serializable;
import java.util.Properties;
/**
* @author yangzheng
* @version $Revision$
* @since 2005-7-14
*/
public class JavaGroupBroadcastingManager
implements NotificationBus.Consumer {
private static final Log log = Log(JavaGroupsBroadcastingListener.class);
private static final String BUS_NAME = "OSCacheBus";
private static final String CHANNEL_PROPERTIES = "cache.cluster.properties";
private static final String MULTICAST_IP_PROPERTY = "cache.cluster.multicast.ip";
private NotificationBus bus;
/
**
* Initializes the broadcasting listener by starting up a JavaGroups notification
* bus instance to handle incoming and outgoing messages.
*
*/
public synchronized void initialize(Properties config) throws InitializationException {
String properties = Property(CHANNEL_PROPERTIES);
String multicastIP = Property(MULTICAST_IP_PROPERTY);
if (log.isInfoEnabled()) {
log.info("Starting a new JavaGroups broadcasting listener with properties="
properties);
}
try {
bus = new NotificationBus(BUS_NAME, properties);
bus.start();
Channel().tOpt(Channel.LOCAL, new Boolean(fal));
bus.tConsumer(this);
log.info("JavaGroups clustering support started successfully");
} catch (Exception e) {
throw new InitializationException("Initialization failed: " e);
}
}
/**
保管协议* Shuts down the JavaGroups being managed
*/
public synchronized void finialize() throws FinalizationException {
if (log.isInfoEnabled()) {
log.info("JavaGroups ");
}
bus.stop();
bus = null;
if (log.isInfoEnabled()) {
log.info("JavaGroups shutdown complete.");
}
言及之而不言}
/**
* Us JavaGroups to broadcast the supplied notification message across the cluster.
*
*/
protected void ndNotification(Serializable message) {
bus.ndNotification(message);
}
/**
* Handles incoming notification messages from JavaGroups. This method should
* never be called directly.
*
*/
public void handleNotification(Serializable rializable) {
log.info("An cluster notification message received message " String()
"). Notification ignored.");
}
/**
* We are not using the caching, so we just return something that identifies
* us. This method should never be called directly.
*/
public Serializable getCache() {
return "JavaGroupsBroadcastingListener: " LocalAddress();
}
/**
* A callback that is fired when a new member joins the cluster. This
* method should never be called directly.
*
* @param address The address of the member who just joined.
*/
public void memberJoined(Address address) {
if (log.isInfoEnabled()) {
log.info("A new member at address '" address "' has joined the cluster");
}
}
/**
* A callback that is fired when an existing member leaves the cluster.
* This method should never be called directly.
*
* @param address The address of the member who left.
*/
public void memberLeft(Address address) {
if (log.isInfoEnabled()) {
log.info("Member at address '" address "' left the cluster");
}
}
}
2、发送消息的程序:
;
import java.io.FileInputStream;
import java.util.Properties;
/**
* @author yangzheng
* @version $Revision$
揭疮疤* @since 2005-7-14
*/
public class TestJavaGroupBroadcastSend {
public static void main(String[] args) throws Exception {
JavaGroupBroadcastingManager javaGroupBroadcastingManager = new JavaGroupBroadcastingManager();
Properties properties = new Properties();
properties.load(new FileInputStream("javagroup.properties"));
javaGroupBroadcastingManager.initialize(properties);
