一：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);
}
}