Memcache MemcacheQ Java 实现生产者消费者

布满荆棘的人生 2022-08-07 11:55 17阅读 0赞

上一节我们将来MemcacheQ安装与使用，这一节我们结合Memcached的java客户端xmemcached实现生产者和消费者消息队列模型。

**1.定义接收消息的回调接口**

package cn.slimsmart.memcache.demo.test.mq;
    
    //接收消息回调
    public interface MemcacheQCallback {
    	//接收消息
    	void receice(Object message);
    }

**2.实现接收消息的客户端**

package cn.slimsmart.memcache.demo.test.mq;
    
    import java.util.concurrent.TimeoutException;
    
    import net.rubyeye.xmemcached.MemcachedClient;
    import net.rubyeye.xmemcached.exception.MemcachedException;
    
    //接收消息客户端
    public class MemcacheQClient implements Runnable {
    
    	private MemcachedClient memcachedClient;
    	private MemcacheQCallback memcacheQCallback;
    	private String queueName;
    
    	public MemcacheQClient() {
    	}
    
    	public MemcacheQClient(MemcachedClient memcachedClient, String queueName, MemcacheQCallback memcacheQCallback) {
    		this.memcachedClient = memcachedClient;
    		this.memcacheQCallback = memcacheQCallback;
    		this.queueName = queueName;
    	}
    
    	@Override
    	public void run() {
    		while (true) {
    			try {
    				Object message = memcachedClient.get(queueName);
    				//如果队列中已经没有数据了,休息一下再试
    				if(message == null){
    					Thread.sleep(10);
    				}else{
    					memcacheQCallback.receice(message);
    				}
    			} catch (TimeoutException e) {
    			} catch (InterruptedException e) {
    			} catch (MemcachedException e) {
    			}
    		}
    	}
    
    	public MemcachedClient getMemcachedClient() {
    		return memcachedClient;
    	}
    
    	public void setMemcachedClient(MemcachedClient memcachedClient) {
    		this.memcachedClient = memcachedClient;
    	}
    
    	public MemcacheQCallback getMemcacheQCallback() {
    		return memcacheQCallback;
    	}
    
    	public void setMemcacheQCallback(MemcacheQCallback memcacheQCallback) {
    		this.memcacheQCallback = memcacheQCallback;
    	}
    
    	public String getQueueName() {
    		return queueName;
    	}
    
    	public void setQueueName(String queueName) {
    		this.queueName = queueName;
    	}
    }

**3.实现生产者，发送消息**

package cn.slimsmart.memcache.demo.test.mq;
    
    import net.rubyeye.xmemcached.XMemcachedClient;
    
    //发送消息
    public class Producer {
    
    	public static void main(String[] args) throws Exception {
    		XMemcachedClient client = new XMemcachedClient("192.168.36.189",22201);
    	    for( int i = 0 ; i < 10 ; i++){
    	    	client.set("queue_test", 0, "hello world - "+i);
    	    	System.out.println("send message : "+"hello world - "+i);
    	    }
    	}
    }

**4.消费者**

package cn.slimsmart.memcache.demo.test.mq;
    
    import net.rubyeye.xmemcached.XMemcachedClient;
    
    //消费
    public class Consumer {
    
    	public static void main(String[] args) throws Exception {
    	    //XMemcachedClient是线程安全的，可以被多线程使用
    	    XMemcachedClient client = new XMemcachedClient("192.168.36.189",22201); 
    	    new Thread(new MemcacheQClient(client, "queue_test", new MemcacheQCallback() {
    			@Override
    			public void receice(Object message) {
    				System.out.println("接收到消息："+message);
    			}
    		})).start();
    	}
    }

运行看下效果吧。。。