资源链接
[Jedis的jar包] 1
[Commons-io的jar包] 2
使用方法
代码样例如下，使用前，注意打开redis的server程序。
代码样例
package RedisExample;
import redis.clients.jedis.Jedis;
public class TestRedis {
public static void main(String[] args) {
Jedis redis = new  Jedis("localhost");
//SimpleExample(redis);
//ListExample(redis,20000);
PublishExample(redis,20000);
}
//简单添加信息
public static void SimpleExample(Jedis redis){
redis.set("key1", "I am value 1");
String ss = redis.get("key1");
System.out.println(ss);
}
//队列添加信息
public static void ListExample(Jedis redis,int number){
String messageStr = "";
int count = 0;
while(count++ < number){
messageStr =  "this is "+count+" message!";
redis.rpush("logstash-test-list",messageStr);
System.out.println(messageStr);
}
}
//发布订阅
public static void PublishExample(Jedis redis,int number){
String messageStr = "";
int count = 0;
while(count++ < number){
messageStr =  "this is "+count+" message!";
redis.publish("logstash-test-list",messageStr);
System.out.println(messageStr);
}
}
}

文章目录
1.消息发布者：实现消息发布的功能
2.消息订阅者：实现消息的订阅消费功能
3.按照上面说明完成基础配置
3.2 在启动2个Redis服务端
3.1在启动2个Redis客户端
3.2 分别在两个客户端中执行以下命令发送消息





1.消息发布者：实现消息发布的功能
package com.ruoyi.project.redis.messgaeQueue;

import redis.clients.jedis.Jedis;

/**
 * 消息发布者
 * 1.建立两个Jedis客户端
 * 2.建立两个发布/订阅监听器
 * 3.启动两个线程。分别用于监听频道和模式
 */
public class MessagePublisher {

    @SuppressWarnings("resource")
    public static void main(String[] args) {
        final Jedis jedis = new Jedis("127.0.0.1", 6379);
        final Jedis pjedis = new Jedis("127.0.0.1", 6380);

        final MessageSubscriber listener = new MessageSubscriber();
        final MessageSubscriber plistener = new MessageSubscriber();

        Thread thread = new Thread(new Runnable() {
            @Override
            public void run() {
                jedis.subscribe(listener, "mychannel");
            }
        });
        Thread pthread = new Thread(new Runnable() {
            @Override
            public void run() {
                pjedis.psubscribe(plistener, "mychannel.*");
            }
        });

        thread.start();
        pthread.start();
    }

}

2.消息订阅者：实现消息的订阅消费功能
package com.ruoyi.project.redis.messgaeQueue;

import redis.clients.jedis.JedisPubSub;

/**
 * 消息订阅者
 * 监听器会对频道和模式的订阅，接收消息和退订等事件进行监听，然后进行相应的处理
 */
public class MessageSubscriber extends JedisPubSub {
    //消息订阅成功的处理
    public void onMessage(String channel, String message) {
        System.out.println("onMessage:" + channel + "=" + message);
        if (message.equals("quit")) {
            this.unsubscribe(channel);
        }
    }

    //订阅初始化
    public void onSubscribe(String channel, int subscribedChannels) {
        System.out.println("onSubscribe:" + channel + "=" + subscribedChannels);
    }

    //取消订阅
    public void onUnsubscribe(String channel, int subscribedChannels) {
        System.out.println("onUnsubscribe:" + channel + "=" + subscribedChannels);
    }

    //按模式的方式订阅
    public void onPSubscribe(String pattern, int subscribedChannels) {
        System.out.println("onPSubscribe:" + pattern + "=" + subscribedChannels);
    }

    //取消模式方式的订阅
    public void onPUnsubscribe(String pattern, int subscribedChannels) {
        System.out.println("onPUnsubscribe:" + pattern + "=" + subscribedChannels);
    }

    //处理消息订阅模式
    public void onPMessage(String pattern, String channel, String message) {
        System.out.println("onPMessage:" + pattern + "=" + channel + "=" + message);
        if (message.equals("quit")) {
            this.punsubscribe(pattern);
        }
    }


}


3.按照上面说明完成基础配置
3.2 在启动2个Redis服务端
完成启动Redis服务后，运行main方法，如下，项目处于消息订阅状态

D:\software\Redis>redis-server D:\software\Redis\redis.windows-6379.conf

D:\software\Redis>redis-server D:\software\Redis\redis.windows-6380.conf






3.1在启动2个Redis客户端

D:>cd D:\software\Redis

D:\software\Redis>redis-cli -p 6379

D:\software\Redis>redis-cli -p 6380


3.2 分别在两个客户端中执行以下命令发送消息

publish mychannel “hello redis”

publish mychannel.redis “hello redis”

前言
现在越来越多的网站开始注重统计和用户行为分析，作为网站经常使用的功能，如何让统计性能更加高，这也是我们需要考虑的事情。本篇通过Redis来优化统计功能(以注册人数统计为例)。
传统的统计功能都是直接操作数据库把数据插入表中。这样做，对数据库的性能消耗就会比较大。
思路：
这里我们用到了redis的队列，注册的时候先添加到队列，然后在处理的时候出队，并且把人数添加redis里。
代码：
//register.php
$redis = new Redis();
$redis->connect('127.0.0.1',6379);
$i=0;
while(true){
$i++;
//假定一直有人在注册
$redis->rpush("register_success",$i);
}
//deal.php
$redis = new Redis();
$redis->connect('127.0.0.1',6379);
while (true) {
//list类型出队操作
$value = $redis->lpop('register_success');
if($value){
echo "deal value : ".$value;
//自增 添加注册人数统计  如果key不存在 则会初始化为0
$redis->incr('register_num');
}else{
echo "deal finish";
}
}
以上这篇PHP+Redis 消息队列 实现高并发下注册人数统计的实例就是小编分享给大家的全部内容了，希望能给大家一个参考，也希望大家多多支持脚本之家。

使用jar包：jedis-2.9.0.jar、commons-pool2-2.3.jar

Message 需要传送的实体类(需实现Serializable接口)
	
ObjectUtil 将对象和byte数组双向转换的工具类
	
Jedis 通过消息队列的先进先出(FIFO)的特点结合Redis的list中的push和pop操作进行封装的工具类
1、消息类

public class Message implements Serializable{
	private int id;
	private String content;

	public Message(int id,String content) {
		this.id = id;
		this.content = content;
	}

	public int getId() {
		return id;
	}
	public void setId(int id) {
		this.id = id;
	}
	public String getContent() {
		return content;
	}
	public void setContent(String content) {
		this.content = content;
	}
}

 

2、序列化

        主要是将对象转化为byte数组,和根据byte数组反序列化成java对象; 主要是用到了ByteArrayOutputStream和ByteArrayInputStream; 注意:每个需要序列化的对象都要实现Serializable接口; 

public class ObjectUtil {
	
	public static byte[] object2Bytes(Object obj) throws IOException{
		ByteArrayOutputStream bo = new ByteArrayOutputStream();
		ObjectOutputStream oo = new ObjectOutputStream(bo);
		oo.writeObject(obj);
		byte[] bytes = bo.toByteArray();
		bo.close();
		oo.close();
		return bytes;
	}
	
	public static Object bytes2Object(byte[] bytes) throws Exception {
		ByteArrayInputStream in = new ByteArrayInputStream(bytes);
		ObjectInputStream sIn = new ObjectInputStream(in);
		return sIn.readObject();
	}
}

3、利用redis做队列,采用redis中list的push和pop操作;

public class JedisUtil {
	private static String JEDIS_IP;
	private static int JEDIS_PORT;
	private static String JEDIS_PASSWORD;
	private static JedisPool jedisPool;
	static{
		Configuration conf = Configuration.getInstance();
		JEDIS_IP="127.0.0.1";
		JEDIS_PORT = 6379;
		JEDIS_PASSWORD = "123456";
		JedisPoolConfig config = new JedisPoolConfig();
	        config.setMaxIdle(256);
	        config.setTestOnBorrow(true);
	        config.setTestOnReturn(true);
	        config.setTestWhileIdle(true);
	        config.setMinEvictableIdleTimeMillis(60000L);
	        config.setTimeBetweenEvictionRunsMillis(3000L);
	        config.setNumTestsPerEvictionRun(-1);
	        jedisPool=new JedisPool(config,JEDIS_IP,JEDIS_PORT,60000,JEDIS_PASSWORD);	
	}
	
    /**
     * 存储REDIS队列 顺序存储
     * @param  key reids键名
     * @param  value 键值
     */
    public static void lpush(byte[] key, byte[] value) {

        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            jedis.lpush(key, value);
        } catch (Exception e) {
            //释放redis对象
            jedisPool.returnBrokenResource(jedis);
            e.printStackTrace();
        } finally {
            //返还到连接池
            close(jedis);
        }
    }

    /**
     * 存储REDIS队列 反向存储
     * @param  key reids键名
     * @param  value 键值
     */
    public static void rpush(byte[] key, byte[] value) {

        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            jedis.rpush(key, value);
        } catch (Exception e) {
            //释放redis对象
            jedisPool.returnBrokenResource(jedis);
            e.printStackTrace();
        } finally {
            //返还到连接池
            close(jedis);
        }
    }

    /**
     * 获取队列数据
     * @param  key 键名
     * @return
     */
    public static byte[] rpop(byte[] key) {

        byte[] bytes = null;
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            bytes = jedis.rpop(key);
        } catch (Exception e) {
            //释放redis对象
            jedisPool.returnBrokenResource(jedis);
            e.printStackTrace();
        } finally {
            //返还到连接池
            close(jedis);
        }
        return bytes;
    }
   
    private static void close(Jedis jedis) {
        try{
            jedisPool.returnResource(jedis);
        }catch (Exception e){
            if(jedis.isConnected()){
                jedis.quit();
                jedis.disconnect();
            }
        }
    }
}

4、测试

用lpush和rpop对redis的队列中的元素进行存入和取出，就能满足先进先出(FIFO)的特点。存入队列中的是Message对象序列化后的字节数组，取出时，再将其反序列化为原对象。

public class TestRedisQueue {

	public static byte[] redisKey = "key".getBytes();
    static {
        try {
            init();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    
    //初始存入1000个消息到队列
    private static void init() throws IOException {
        for (int i = 0; i < 1000; i++) {
            Message message = new Message(i, "这是第" + i + "个内容");
            JedisUtil.lpush(redisKey, ObjectUtil.object2Bytes(message));
        }
    }

    //每运行一次取出100个消息
    public static void main(String[] args) {
        try {
        	for(int i=0;i<100;i++){
        		pop();
        	}            
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static void pop() throws Exception {
        byte[] bytes = JedisUtil.rpop(redisKey);
        Message msg = (Message) ObjectUtil.bytes2Object(bytes);
        if (msg != null) {
            System.out.println(msg.getId() + "----" + msg.getContent());
        }
    }
}