一：Redis的使用
1.先引入pom.xml的依赖
<dependency> <groupId>redis.clients</groupId> <artifactId>jedis</artifactId> <version>2.9.0</version> </dependency>
2.开始编码
package me.silentdoer.redisrabbitmq; import redis.clients.jedis.Jedis; import redis.clients.jedis.JedisPool; import redis.clients.jedis.JedisPoolConfig; import redis.clients.util.Pool; import java.io.UnsupportedEncodingException; /** * @author silentdoer * @version 1.0 * @description the description * @date 4/11/18 9:03 PM */ public class Entrance { public static Pool<Jedis> jedisPool;
// TODO 在项目里对redis的使用其实更通过radis-cli的方式几乎一模一样； public static void main(String[] args) throws UnsupportedEncodingException { // 通过配置初始化Jedis连接池，类似初始化数据库连接池的步骤 initJedisPool(); // 通过Jedis连接池来管理Jedis，这样当jedis.close()时能够复用内部的数据传输模块 Jedis jedis = jedisPool.getResource(); // 设置db0中的key，value jedis.set("中国".getBytes("utf8"), "我里试试试".getBytes("utf8")); // 对于set字符串，最终也是通过set(SafeEncoder.encode(key), SafeEncoder.encode(value))转换为byte数组,看了下是UTF-8 //jedis.set("strKey", "value"); // 这里的value最终也是通过SafeEncoder.encodeMany(strs)，将字符串数组（List类型的value）转换为了二位字节数组 //jedis.rpush("listKey", new String[]{"value1", "value2", "value3"}); // 注意，set时是什么形式，get时最好就应该是什么形式 byte[] content = jedis.get("中国".getBytes("utf8")); System.out.println(new String(content, "UTF8")); // 虽然这样也行，但是为了未来兼容性，还是手动转换为字节数组比较靠谱，比较后面的版本不一定就换成别的编码了 //String content = jedis.get("中国"); //System.out.println(content); //选择redis的内存db，这个大小是可以在配置里规定的，个人规定的是一个redis实例的db大小不能超过16 //jedis.select(1); // 若此Jedis对象由JedisPool创建则close时将会设置此jedis不可用，但是回收内部的数据传输模块 jedis.close(); } public static synchronized void initJedisPool(){ JedisPoolConfig config = new JedisPoolConfig(); config.setMaxIdle(24); // 最大空闲数（这里应该还有个设置多久后认定是空闲的） config.setMaxTotal(48); // 最大连接数 config.setMaxWaitMillis(3000); // 当连接不够时等待的时长 config.setTestOnBorrow(false); try { jedisPool = new JedisPool(config, "localhost", 6379, 10000, null); // 最后一个为auth，没有即为null } catch (Exception ex) { ex.printStackTrace(); } } }
二：RabbitMQ的使用
1.引入依赖
<dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>4.1.0</version> </dependency>
2.生产者编码
package me.silentdoer.redisrabbitmq; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import java.io.IOException; import java.util.Scanner; import java.util.concurrent.TimeoutException; /** * @author silentdoer * @version 1.0 * @description the description * @date 4/11/18 10:23 PM */ public class MqProducer { private static ConnectionFactory factory; public static void main(String[] args) throws IOException, TimeoutException { initFactory(); Connection connection = factory.newConnection(); // channel的实现方式其实就是注册机制，即我通过某个tcp协议的socket对象，调有createChannel其实就是发送如：create channel 0的数据 // 然后服务端收到后在相关注册表里添加条目 0 - conn0 - user - vhost 之类的数据，然后返回true给客户端，然后客户端也添加本地注册表 // 这样就通过每次发送数据都加一个channel从而将不同channel的数据分隔开 Channel chann0 = connection.createChannel(1); // TODO 此值要大于0 Scanner scanner = new Scanner(System.in); String msg = ""; while(!"exit".equals(msg)){ msg = scanner.nextLine(); chann0.basicPublish("", "test", null, msg.getBytes("utf8")); } // ""表示是默认的exchange(direct.default)，但是还是得显示指定，TODO 注意不能是null //第三个参数props是消息包含的属性信息。RabbitMQ的消息属性和消息体是分开的，不像JMS消息那样同时包含在javax.jms.Message对象中，这一点需要特别注意 //chann0.basicPublish("", "test", null, "payload的消息".getBytes("utf8")); //chann0.queueDeclare("name", true, false, false, null); // 创建一个queue（默认会创建一个default-queuename-queue的binding） //chann0.exchangeDeclare(..); // 创建一个exchange，只能指定builtin的类型，即direct/fanout/header/topic。。 //chann0.exchangeBind(..); // 设置一个binding，将exchange和exchange绑定，类似责任链处理 //chann0.queueBind(..); // 设置一个binding，将queue和exchange绑定 chann0.close(); // 注意，其实只是在本地和服务器的注册表里删除了此channel的信息，以及将此channel设置为不可用而以 connection.close(); // 真正关闭tcp连接 } public static void initFactory(){ factory = new ConnectionFactory(); factory.setHost("localhost"); factory.setPort(5672); factory.setUsername("guest"); factory.setPassword("guest"); factory.setVirtualHost("/"); // default vhost is / factory.setConnectionTimeout(10000); } }
3.消费者代码
package me.silentdoer.redisrabbitmq; import com.rabbitmq.client.*; import java.io.IOException; import java.util.concurrent.TimeoutException; /** * @author silentdoer * @version 1.0 * @description the description * @date 4/11/18 10:43 PM */ public class MqConsumer { private static ConnectionFactory factory; public static void main(String[] args) throws IOException, TimeoutException { initFactory(); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(3); // channel 不需要和producer的一样 // Consumer 是回调函数，即channel.basicConsume(..)获得一条消息后会调有此观察者的 //handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)方法 Consumer consumer = new DefaultConsumer(channel){ // 默认此观察者的命令方法是什么都不做的，要将这里获取的值在其它地方引用只需写个ConsumerSupport即可，然后外部consumerSupport.getBody(); @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String message = new String(body, "UTF-8"); System.out.println("收到一条需要我方处理的消息:" + message); } }; while (true){ // 注意，是channel在订阅消息，这个订阅和监听器不太一样，需要不断的订阅（有点像C#里的BeginAccept，获取后需要继续BeginAccept才行） channel.basicConsume("test", true, consumer); }
//channel.close();
//connection.close(); } public static void initFactory(){ factory = new ConnectionFactory(); factory.setHost("localhost"); factory.setPort(5672); factory.setUsername("guest"); // TODO 注意，生产环境消费者和生产者的账户是不一样的，只不过vhost一样，ip:port也一样 factory.setPassword("guest"); factory.setVirtualHost("/"); factory.setConnectionTimeout(10000); } }