OPENTECH / petstore · Commits

Go to a project

Authored by jack.lee
Commit 8e60c1068de8cbee5ff0767fe0e024c27eecdca3 1 parent 3190be90

no message

Inline Side-by-side
Showing 33 changed files with 4059 additions and 213 deletions
... ... @@ -18,5 +18,10 @@
<groupId>javax.servlet</groupId>
<artifactId>servlet-api</artifactId>
</dependency>
<dependency>
<groupId>org.elasticsearch</groupId>
<artifactId>elasticsearch</artifactId>
<version>1.4.2</version>
</dependency>
</dependencies>
</project>
... ...
/**
* Project: petstore-common
*
* File Created at 2015年10月29日
* $Id$
*
* Copyright 2011 Yoho.cn Croporation Limited.
* All rights reserved.
*
* This software is the confidential and proprietary information of
* Yoho Company. ("Confidential Information"). You shall not
* disclose such Confidential Information and shall use it only in
* accordance with the terms of the license agreement you entered into
* with Yoho.cn.
*/
package com.taobao.sample.common;
import java.util.concurrent.Callable;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import com.taobao.sample.common.concurrent.ThreadHelper;
/**
* 方法异步调用拦截器
*
* @author Jack.lee
* @version $Id: MethodAsyncInterceptor.java 2015年10月29日 下午1:16:29 $
*
*/
public class MethodAsyncInterceptor implements MethodInterceptor {
ThreadHelper threadHelper;
/**
* @param threadHelper the threadHelper to set
*/
public void setThreadHelper(ThreadHelper threadHelper) {
this.threadHelper = threadHelper;
}
@Override
public Object invoke(MethodInvocation invocation) throws Throwable {
Object result = threadHelper.execute(new Callable<Object>() {
@Override
public Object call() throws Exception {
try {
return invocation.proceed();
} catch (Throwable e) {
e.printStackTrace();
}
return null;
}
});
return result;
}
}
... ...
/**
* Project: petstore-common
*
* File Created at 2015年10月26日
* $Id$
*
* Copyright 2011 Yoho.cn Croporation Limited.
* All rights reserved.
*
* This software is the confidential and proprietary information of
* Yoho Company. ("Confidential Information"). You shall not
* disclose such Confidential Information and shall use it only in
* accordance with the terms of the license agreement you entered into
* with Yoho.cn.
*/
package com.taobao.sample.common;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import com.taobao.common.lang.diagnostic.Profiler;
/**
* TODO Comment of MethodProfilerInterceptor
*
* @author Jack.lee
* @version $Id: MethodProfilerInterceptor.java 2015年10月26日 下午7:49:28 $
*
*/
public class MethodProfilerInterceptor implements MethodInterceptor {
private boolean showArguments = false;
/**
* @param showArguments the showArguments to set
*/
public void setShowArguments(boolean showArguments) {
this.showArguments = showArguments;
}
@Override
public Object invoke(MethodInvocation invocation) throws Throwable {
Profiler.enter(getMethod(invocation));
Object result = invocation.proceed();
Profiler.release();
return result;
}
private String getMethod(MethodInvocation invocation) {
String targetName = invocation.getThis().getClass().getName();
String methodName = invocation.getMethod().getName();
Object[] arguments = invocation.getArguments();
StringBuffer sb = new StringBuffer();
sb.append(targetName).append(".").append(methodName);
if (showArguments) {
sb.append("(");
if ((arguments != null) && (arguments.length != 0)) {
for (int i = 0; i < arguments.length; i++) {
if (i > 0) {
sb.append(",");
}
sb.append(String.valueOf(arguments[i]));
}
}
sb.append(")");
}
return sb.toString();
}
}
... ...
/**
* Project: sc-common
*
* File Created at 2011-11-1
* $Id$
*
* Copyright 2011 Milanoo.com Croporation Limited.
* All rights reserved.
*
* This software is the confidential and proprietary information of
* Milanoo Company. ("Confidential Information"). You shall not
* disclose such Confidential Information and shall use it only in
* accordance with the terms of the license agreement you entered into
* with Milanoo.com.
*/
package com.taobao.sample.common.concurrent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* TODO Comment of AbstractRunable
*
* @author Lijian
* @version $Id: AbstractRunable.java 2011-11-1 上午11:26:00 $
*/
public abstract class AbstractRunable implements Runnable {
/** logger */
protected final Logger log = LoggerFactory.getLogger(getClass());
private TaskExecutor exec;
public AbstractRunable(TaskExecutor exec) {
this.exec = exec;
}
/*
* (non-Javadoc)
* @see java.lang.Runnable#run()
*/
@Override
public void run() {
try {
execute();
} finally {
exec.getSemaphore().release();
}
}
protected abstract void execute();
}
... ...
/**
* Project: sc-common
*
* File Created at 2011-11-1
* $Id$
*
* Copyright 2011 Milanoo.com Croporation Limited.
* All rights reserved.
*
* This software is the confidential and proprietary information of
* Milanoo Company. ("Confidential Information"). You shall not
* disclose such Confidential Information and shall use it only in
* accordance with the terms of the license agreement you entered into
* with Milanoo.com.
*/
package com.taobao.sample.common.concurrent;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.Semaphore;
/**
* 任务执行线程池
*
* @author Lijian
* @version $Id: TaskExecutor.java 2011-11-1 上午11:16:58 $
*/
public class TaskExecutor {
/** 最大并发数 */
private int taskExecPoolMaxCount = 500;
/** 用于提交任务的线程池 */
private final ExecutorService execPool;
/** 阻塞信号量 */
private final Semaphore semaphore;
/**
* @param taskExecPoolMaxCount the taskExecPoolMaxCount to set
*/
public void setTaskExecPoolMaxCount(int taskExecPoolMaxCount) {
this.taskExecPoolMaxCount = taskExecPoolMaxCount;
}
public TaskExecutor() {
this.execPool = Executors.newFixedThreadPool(taskExecPoolMaxCount);
this.semaphore = new Semaphore(taskExecPoolMaxCount);
}
/**
* 获取信号量
*
* @return
*/
public Semaphore getSemaphore() {
return semaphore;
}
/**
* 提交任务
*
* @param runnable
* @return
* @throws InterruptedException
*/
public Future<?> submit(Runnable runnable) throws InterruptedException {
semaphore.acquire();
return execPool.submit(runnable);
}
/**
* 关闭线程池
*/
public void destory() {
execPool.shutdownNow();
}
}
... ...
/**
* Project: sc-common
*
* File Created at 2011-11-1
* $Id$
*
* Copyright 2011 Milanoo.com Croporation Limited.
* All rights reserved.
*
* This software is the confidential and proprietary information of
* Milanoo Company. ("Confidential Information"). You shall not
* disclose such Confidential Information and shall use it only in
* accordance with the terms of the license agreement you entered into
* with Milanoo.com.
*/
package com.taobao.sample.common.concurrent;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
/**
* 任务队列
*
* @author Lijian
* @version $Id: TaskQueue.java 2011-11-1 上午11:11:30 $
*/
public class TaskQueue {
/** 队列容量 */
private int taskqueueMaxCount = 500;
/** 用于多线程间存取任务的队列，支持优先级，任务对象需要实现Comparable接口 */
private BlockingQueue<Object> queue = null;
/**
* @param taskqueueMaxCount the taskqueueMaxCount to set
*/
public void setTaskqueueMaxCount(int taskqueueMaxCount) {
this.taskqueueMaxCount = taskqueueMaxCount;
}
public TaskQueue() {
queue = new LinkedBlockingQueue<Object>(taskqueueMaxCount);
}
/**
* 入队列 如果队列已满则阻塞
*
* @param task
* @throws InterruptedException
*/
public void put(Object task) throws InterruptedException {
queue.put(task);
}
/**
* 出队列 如果队列为空则阻塞
*
* @return
* @throws InterruptedException
*/
public Object take() throws InterruptedException {
return queue.take();
}
/**
* 删除队列数据
*
* @return
*/
public Object remove() {
return queue.remove();
}
/**
* 删除指定对象
*
* @param obj
* @return
*/
public boolean remove(Object task) {
return queue.remove(task);
}
/**
* 清空任务队列
*/
public void clear() {
queue.clear();
}
}
... ...
/**
* Project: sc-common
*
* File Created at 2011-11-1
* $Id$
*
* Copyright 2011 Milanoo.com Croporation Limited.
* All rights reserved.
*
* This software is the confidential and proprietary information of
* Milanoo Company. ("Confidential Information"). You shall not
* disclose such Confidential Information and shall use it only in
* accordance with the terms of the license agreement you entered into
* with Milanoo.com.
*/
package com.taobao.sample.common.concurrent;
/**
* <p>
* 回调接口
* </p>
* <p>
* 通过实现该接口完成业务逻辑
* </p>
*
* @author Lijian
* @version $Id: ThreadCallback.java 2011-11-1 下午10:48:34 $
*/
public interface ThreadCallback {
void run();
}
... ...
/**
* Project: sc-common
*
* File Created at 2011-11-1
* $Id$
*
* Copyright 2011 Milanoo.com Croporation Limited.
* All rights reserved.
*
* This software is the confidential and proprietary information of
* Milanoo Company. ("Confidential Information"). You shall not
* disclose such Confidential Information and shall use it only in
* accordance with the terms of the license agreement you entered into
* with Milanoo.com.
*/
package com.taobao.sample.common.concurrent;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* <p>
* 线程助手类
* </p>
* <p>
* 通过回调的方式完成业务代码的异步调用
* </p>
*
* @author Lijian
* @version $Id: ThreadHelper.java 2011-11-1 下午10:47:45 $
*/
public class ThreadHelper {
/** logger */
protected final Logger log = LoggerFactory.getLogger(getClass());
private TaskExecutor taskExecutor;
/**
* @param taskExecutor the taskExecutor to set
*/
public void setTaskExecutor(TaskExecutor taskExecutor) {
this.taskExecutor = taskExecutor;
}
@SuppressWarnings({ "unchecked", "rawtypes" })
public Future<?> execute(Callable<?> task) {
FutureTask<?> futureTask = new FutureTask(task);
try {
taskExecutor.submit(futureTask);
} catch (InterruptedException e) {
log.error("Run thread failed.", e);
}
return futureTask;
}
/**
* 通过回调执行业务方法
*
* @param callback 回调接口
*/
public Future<?> execute(ThreadCallback callback) {
TaskRunable tr = new TaskRunable(taskExecutor, callback);
try {
return taskExecutor.submit(tr);
} catch (InterruptedException e) {
log.error("Run thread failed.", e);
return null;
}
}
/**
* 通过回调调度业务方法
*
* @param interval 调度间隔时间
* @param callback 回调接口
*/
public ScheduleThread execute(long interval, ThreadCallback callback) {
ScheduleThread st = new ScheduleThread(callback, interval);
st.start();
return st;
}
/**
* 任务执行线程
*
* @author Lijian
* @version $Id: ThreadHelper.java 2011-11-1 下午11:16:03 $
*/
protected class TaskRunable extends AbstractRunable {
ThreadCallback callback;
/**
* @param exec
*/
protected TaskRunable(TaskExecutor exec, ThreadCallback callback) {
super(exec);
this.callback = callback;
}
@Override
protected void execute() {
callback.run();
}
}
/**
* 任务调度线程
*
* @author Lijian
* @version $Id: ThreadHelper.java 2011-11-1 下午11:32:29 $
*/
protected class ScheduleThread extends Thread {
private boolean stopThread = false;
private long interval = 1000 * 5; // every 5 seconds
ThreadCallback callback;
protected ScheduleThread(ThreadCallback callback, long interval) {
this.callback = callback;
this.interval = interval;
}
/**
* sets stop variable and interupts any wait
*/
public void stopThread() {
this.stopThread = true;
this.interrupt();
}
/**
* Start the thread.
*/
public void run() {
while (!this.stopThread) {
try {
Thread.sleep(interval);
callback.run();
} catch (InterruptedException e) {
if (e instanceof java.lang.InterruptedException)
log.info("ScheduleThread stop !");
else
log.error("ScheduleThread error !", e);
break;
}
}
}
}
}
... ...
<?xml version="1.0" encoding="UTF-8" ?>
<beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns="http://www.springframework.org/schema/beans"
xmlns:p="http://www.springframework.org/schema/p"
xsi:schemaLocation="
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.1.xsd"
default-autowire="byName">
<bean id="autoProxyCreator" class="org.springframework.aop.framework.autoproxy.DefaultAdvisorAutoProxyCreator"/>
<!-- ===================================================================== -->
<!-- 记录业务方法的执行时间 -->
<!-- ===================================================================== -->
<bean id="serviceTimerAdvisor" class="org.springframework.aop.support.RegexpMethodPointcutAdvisor">
<property name="advice" ref="methodProfilerInterceptor"/>
<property name="patterns">
<list>
<value>com.taobao.sample.petstore.biz.store.service\..+</value>
<value>com.taobao.sample.petstore.biz.user.service\..+</value>
</list>
</property>
</bean>
<bean id="methodProfilerInterceptor" class="com.taobao.sample.common.MethodProfilerInterceptor" >
<property name="showArguments" value="true"/> <!-- 是否记录运行时方法的参数 -->
</bean>
</beans>
\ No newline at end of file
... ...
... ... @@ -47,6 +47,11 @@
<artifactId>mybatis</artifactId>
<version>3.2.7</version>
</dependency>
<dependency>
<groupId>com.yoho.sample</groupId>
<artifactId>${project-name}-common</artifactId>
<version>${project-version}</version>
</dependency>
</dependencies>
<build>
... ...
/**
* Project: petstore-dal
*
* File Created at 2015年10月29日
* $Id$
*
* Copyright 2011 Yoho.cn Croporation Limited.
* All rights reserved.
*
* This software is the confidential and proprietary information of
* Yoho Company. ("Confidential Information"). You shall not
* disclose such Confidential Information and shall use it only in
* accordance with the terms of the license agreement you entered into
* with Yoho.cn.
*/
package com.taobao.sample.petstore.dal.dao;
import java.io.IOException;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import com.alibaba.fastjson.JSONObject;
/**
* TODO Comment of ESClientDemo
*
* @author ibm
* @version $Id: ESClientDemo.java 2015年10月29日 下午5:03:59 $
*/
public class ESClientDemo {
private Client client;
@SuppressWarnings("resource")
public void init() {
client = new TransportClient().addTransportAddress(new InetSocketTransportAddress("127.0.0.1", 9300));
}
public void close() {
client.close();
}
/**
* index
*/
public void createIndex() {
for (int i = 1500; i < 1600; i++) {
User user = new User();
user.setId(new Long(i));
user.setName("wen fox " + i);
user.setAge(i % 100);
client.prepareIndex("users", "user").setSource(JSONObject.toJSONString(user)).execute().actionGet();
}
}
public void createIndex1() {
Student stu = new Student("1", "1", "1");
client.prepareIndex("school", "student").setSource(JSONObject.toJSONString(stu)).execute().actionGet();
stu = new Student("1", "1", "2");
client.prepareIndex("school", "student").setSource(JSONObject.toJSONString(stu)).execute().actionGet();
stu = new Student("1", "2", "3");
client.prepareIndex("school", "student").setSource(JSONObject.toJSONString(stu)).execute().actionGet();
stu = new Student("1", "2", "4");
client.prepareIndex("school", "student").setSource(JSONObject.toJSONString(stu)).execute().actionGet();
stu = new Student("1", "2", "5");
client.prepareIndex("school", "student").setSource(JSONObject.toJSONString(stu)).execute().actionGet();
}
/**
* 转换成json对象
*
* @param user
* @return
*/
private String generateJson(User user) {
String json = "";
try {
XContentBuilder contentBuilder = XContentFactory.jsonBuilder().startObject();
contentBuilder.field("id", user.getId() + "");
contentBuilder.field("name", user.getName());
contentBuilder.field("age", user.getAge() + "");
json = contentBuilder.endObject().string();
} catch (IOException e) {
e.printStackTrace();
}
return json;
}
public static void main(String[] args) {
ESClientDemo client = new ESClientDemo();
client.init();
client.createIndex1();
client.close();
}
}
... ...
/**
* Project: petstore-dal
*
* File Created at 2015年10月29日
* $Id$
*
* Copyright 2011 Yoho.cn Croporation Limited.
* All rights reserved.
*
* This software is the confidential and proprietary information of
* Yoho Company. ("Confidential Information"). You shall not
* disclose such Confidential Information and shall use it only in
* accordance with the terms of the license agreement you entered into
* with Yoho.cn.
*/
package com.taobao.sample.petstore.dal.dao;
import java.util.Iterator;
import java.util.Map;
import org.elasticsearch.action.ActionFuture;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.Requests;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.ImmutableSettings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.index.query.BoolFilterBuilder;
import org.elasticsearch.index.query.FilterBuilders;
import org.elasticsearch.index.query.MatchAllQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.AbstractAggregationBuilder;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.terms.DoubleTerms;
import org.elasticsearch.search.aggregations.bucket.terms.LongTerms;
import org.elasticsearch.search.aggregations.bucket.terms.StringTerms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms.Bucket;
import org.elasticsearch.search.aggregations.bucket.terms.TermsBuilder;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import com.alibaba.fastjson.JSONObject;
/**
* TODO Comment of ESSearchDemo
*
* @author ibm
* @version $Id: ESSearchDemo.java 2015年10月29日 下午11:34:43 $
*
*/
public class ESSearchDemo {
private int port = 9300;
private String ipAddress[] = {"182.92.99.119", "127.0.0.1"};
private static Client client;
public ESSearchDemo(){
Settings settings = ImmutableSettings.settingsBuilder().put("cluster.name", "yohosearch_test").build();
TransportClient transportClient = new TransportClient(settings);
for (String ip : ipAddress) {
transportClient.addTransportAddress(new InetSocketTransportAddress(ip, port));
}
client = transportClient;
}
public static void main(String[] args) {
ESSearchDemo ess = new ESSearchDemo();
testCategoryAggregations();
}
public static void testCategoryAggregations() {
String index = "yoho_productindex_1446693760005_20151105112240013";
String type = "productindex";
BoolFilterBuilder boolFilter = FilterBuilders.boolFilter().must(FilterBuilders.termFilter("maxSortId", 7));
//MultiMatchQueryBuilder query = QueryBuilders.multiMatchQuery("钥匙", "productName");
MatchAllQueryBuilder query = QueryBuilders.matchAllQuery();
QueryBuilder queryBuilder = QueryBuilders.filteredQuery(query, boolFilter);
TermsBuilder aggregation1 = AggregationBuilders.terms("maxSortNameAgg").field("maxSortName").size(1000);
TermsBuilder aggregation2 = AggregationBuilders.terms("middleSortNameAgg").field("middleSortName").size(1000);
TermsBuilder aggregation3 = AggregationBuilders.terms("smallSortNameAgg").field("smallSortName").size(1000);
TermsBuilder aggregation11 = AggregationBuilders.terms("colorIdsAgg").field("colorIds").size(1000);
TermsBuilder aggregation12 = AggregationBuilders.terms("priceAgg").field("salesPrice").size(1000);
aggregation1.subAggregation(aggregation2);
aggregation2.subAggregation(aggregation3);
SearchSourceBuilder sourceBuilder = SearchSourceBuilder.searchSource();
//sourceBuilder.postFilter(boolFilter);
sourceBuilder.query(queryBuilder);
sourceBuilder.aggregation(aggregation1);
sourceBuilder.aggregation(aggregation11);
sourceBuilder.aggregation(aggregation12);
sourceBuilder.size(1);
sourceBuilder.from(1);
SearchRequest request = Requests.searchRequest(index);
request.types(type);
request.source(sourceBuilder.toString());
request.searchType(SearchType.COUNT);
ActionFuture<SearchResponse> res = client.search(request);
SearchResponse searchResponse = res.actionGet();
SearchHits hits = searchResponse.getHits();
System.out.println("查询到记录数=" + hits.getTotalHits());
Map<String, Aggregation> aggMap = searchResponse.getAggregations().asMap();
StringTerms stringTerms1 = (StringTerms) aggMap.get("maxSortNameAgg");
Iterator<Bucket> gradeBucketIt1 = stringTerms1.getBuckets().iterator();
while (gradeBucketIt1.hasNext()) {
Bucket gradeBucket1 = gradeBucketIt1.next();
System.out.println(gradeBucket1.getKey() + "(" + gradeBucket1.getDocCount() + ")");
StringTerms stringTerms2 = (StringTerms) gradeBucket1.getAggregations().asMap().get("middleSortNameAgg");
Iterator<Bucket> gradeBucketIt2 = stringTerms2.getBuckets().iterator();
while (gradeBucketIt2.hasNext()) {
Bucket gradeBucket2 = gradeBucketIt2.next();
System.out.println(">>>" + gradeBucket2.getKey() + "(" + gradeBucket2.getDocCount() + ")");
StringTerms stringTerms3 = (StringTerms) gradeBucket2.getAggregations().asMap().get("smallSortNameAgg");
Iterator<Bucket> gradeBucketIt3 = stringTerms3.getBuckets().iterator();
while (gradeBucketIt3.hasNext()) {
Bucket gradeBucket3 = gradeBucketIt3.next();
System.out.println(">>>>>>" + gradeBucket3.getKey() + "(" + gradeBucket3.getDocCount() + ")");
}
}
}
System.out.println("==================================================");
StringTerms styleAgg = (StringTerms) aggMap.get("colorIdsAgg");
Iterator<Bucket> styleBucketIt = styleAgg.getBuckets().iterator();
while (styleBucketIt.hasNext()) {
Bucket bucketIt = styleBucketIt.next();
System.out.println(bucketIt.getKey() + "(" + bucketIt.getDocCount() + ")");
}
System.out.println("==================================================");
DoubleTerms priceAgg = (DoubleTerms) aggMap.get("priceAgg");
Iterator<Bucket> priceBucketIt = priceAgg.getBuckets().iterator();
while (priceBucketIt.hasNext()) {
Bucket bucketIt = priceBucketIt.next();
System.out.println(bucketIt.getKey() + "(" + bucketIt.getDocCount() + ")");
}
}
public static void testAggregations() {
String index = "yoho_productindex_1446693760005_20151105112240013";
String type = "productindex";
BoolFilterBuilder boolFilter = FilterBuilders.boolFilter().must(FilterBuilders.termFilter("maxSortId", 3));
AbstractAggregationBuilder aggregation = AggregationBuilders.terms("brandAgg").field("brandId").size(1000);
SearchSourceBuilder sourceBuilder = SearchSourceBuilder.searchSource();
sourceBuilder.postFilter(boolFilter);
sourceBuilder.aggregation(aggregation);
sourceBuilder.size(1);
sourceBuilder.from(1);
SearchRequest request = Requests.searchRequest(index);
request.types(type);
request.source(sourceBuilder.toString());
request.searchType(SearchType.COUNT);
ActionFuture<SearchResponse> res = client.search(request);
SearchResponse searchResponse = res.actionGet();
SearchHits hits = searchResponse.getHits();
System.out.println("查询到记录数=" + hits.getTotalHits());
Map<String, Aggregation> aggMap = searchResponse.getAggregations().asMap();
LongTerms longTerms = (LongTerms) aggMap.get("brandAgg");
Iterator<Bucket> gradeBucketIt = longTerms.getBuckets().iterator();
while (gradeBucketIt.hasNext()) {
Bucket gradeBucket = gradeBucketIt.next();
System.out.println(gradeBucket.getKey() + ":" + gradeBucket.getDocCount());
}
}
public static void testQuery() {
String index = "yoho_productindex_1446693760005_20151105112240013";
String type = "productindex";
//QueryBuilder queryBuilder = QueryBuilders.inQuery("productSkn", "51051136");
BoolFilterBuilder boolFilter = FilterBuilders.boolFilter().must(FilterBuilders.termFilter("maxSortId", 3));
//boolFilter.must(FilterBuilders.termFilter("brandId", 141));
//SearchResponse searchResponse = client.prepareSearch(index).setTypes(type).setQuery(queryBuilder).execute().actionGet();
SearchSourceBuilder sourceBuilder = SearchSourceBuilder.searchSource();
sourceBuilder.postFilter(boolFilter);
sourceBuilder.size(1000);
SearchRequest request = Requests.searchRequest(index);
request.types(type);
request.source(sourceBuilder.toString());
ActionFuture<SearchResponse> res = client.search(request);
SearchResponse searchResponse = res.actionGet();
SearchHits hits = searchResponse.getHits();
System.out.println("查询到记录数=" + hits.getTotalHits());
SearchHit[] searchHists = hits.getHits();
if(searchHists.length>0){
for(SearchHit hit:searchHists){
//String productSkn = String.valueOf(hit.getSource().get("productSkn"));
//String productName = String.valueOf(hit.getSource().get("productName"));
//System.out.println(">>>productId:" + productId + "||productSkn:" + productSkn + "||productName:" + productName);
String str = JSONObject.toJSONString(hit.getSource());
System.out.println(str);
}
}
}
}
... ...
/**
* Project: petstore-dal
*
* File Created at 2015年10月29日
* $Id$
*
* Copyright 2011 Yoho.cn Croporation Limited.
* All rights reserved.
*
* This software is the confidential and proprietary information of
* Yoho Company. ("Confidential Information"). You shall not
* disclose such Confidential Information and shall use it only in
* accordance with the terms of the license agreement you entered into
* with Yoho.cn.
*/
package com.taobao.sample.petstore.dal.dao;
import java.util.ArrayList;
import java.util.List;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import com.alibaba.fastjson.JSONObject;
/**
* TODO Comment of ElasticSearchHandler
*
* @author ibm
* @version $Id: ElasticSearchHandler.java 2015年10月29日 下午7:35:54 $
*/
public class ElasticSearchHandler {
private Client client;
public ElasticSearchHandler() {
//使用本机做为节点
this("127.0.0.1");
}
public ElasticSearchHandler(String ipAddress) {
//集群连接超时设置
/*
* Settings settings = ImmutableSettings.settingsBuilder().put(
* "client.transport.ping_timeout", "10s").build(); client = new
* TransportClient(settings);
*/
client = new TransportClient().addTransportAddress(new InetSocketTransportAddress(ipAddress, 9300));
}
/**
* 建立索引,索引建立好之后,会在elasticsearch-0.20.6\data\elasticsearch\nodes\0创建所以你看
*
* @param indexName 为索引库名，一个es集群中可以有多个索引库。 名称必须为小写
* @param indexType Type为索引类型，是用来区分同索引库下不同类型的数据的，一个索引库下可以有多个索引类型。
* @param jsondata json格式的数据集合
* @return
*/
public void createIndexResponse(String indexname, String type, List<String> jsondata) {
//创建索引库 需要注意的是.setRefresh(true)这里一定要设置,否则第一次建立索引查找不到数据
IndexRequestBuilder requestBuilder = client.prepareIndex(indexname, type).setRefresh(true);
for (int i = 0; i < jsondata.size(); i++) {
requestBuilder.setSource(jsondata.get(i)).execute().actionGet();
}
}
/**
* 创建索引
*
* @param client
* @param jsondata
* @return
*/
public IndexResponse createIndexResponse(String indexname, String type, String jsondata) {
IndexResponse response = client.prepareIndex(indexname, type).setSource(jsondata).execute().actionGet();
return response;
}
/**
* 执行搜索
*
* @param queryBuilder
* @param indexname
* @param type
* @return
*/
public List<User> searcher(QueryBuilder queryBuilder, String indexname, String type) {
List<User> list = new ArrayList<User>();
SearchResponse searchResponse = client.prepareSearch(indexname).setTypes(type).setQuery(queryBuilder).setFrom(0).setSize(5).execute().actionGet();
SearchHits hits = searchResponse.getHits();
System.out.println("查询到记录数=" + hits.getTotalHits());
if (hits.getTotalHits() > 0) {
for (SearchHit hit : hits) {
String json = JSONObject.toJSONString(hit.getSource());
System.out.println(">>>json:" + json);
Long id = Long.valueOf(String.valueOf(hit.getSource().get("id")));
String name = (String) hit.getSource().get("name");
Integer age = Integer.valueOf((String) hit.getSource().get("age"));
list.add(new User(id, name, age));
}
}
return list;
}
public static void main(String[] args) {
ElasticSearchHandler esHandler = new ElasticSearchHandler();
String indexname = "users";
String type = "user";
//查询条件
//QueryBuilder queryBuilder = QueryBuilders.inQuery("name","huang");
QueryBuilder queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.termQuery("id", 140));
List<User> result = esHandler.searcher(queryBuilder, indexname, type);
for (int i = 0; i < result.size(); i++) {
User user = result.get(i);
System.out.println(user.getId() + ") name:" + user.getName() + "||age:" + user.getAge());
}
esHandler.client.close();
}
}
... ...
/**
* Project: petstore-dal
*
* File Created at 2015年10月29日
* $Id$
*
* Copyright 2011 Yoho.cn Croporation Limited.
* All rights reserved.
*
* This software is the confidential and proprietary information of
* Yoho Company. ("Confidential Information"). You shall not
* disclose such Confidential Information and shall use it only in
* accordance with the terms of the license agreement you entered into
* with Yoho.cn.
*/
package com.taobao.sample.petstore.dal.dao;
import java.util.List;
import java.util.concurrent.Future;
import com.taobao.sample.petstore.dal.page.PageResult;
/**
* TODO Comment of GenericAsyncDao
*
* @author ibm
* @version $Id: GenericAsyncDao.java 2015年10月29日 下午2:23:19 $
*
*/
public interface GenericAsyncDao {
public static final String DEFAULT_BIZ = "__defaultBiz";
public static final String DELETE = "";//"DELETE_";
public static final int DELETE_OP = 4;
public static final String INSERT = "";//"INSERT_";
public static final int INSERT_OP = 1;
public static final String SELECT = "";//"SELECT_";
public static final String SELECT_BY_PK = "SELECT_BY_PK";
public static final String SELECT_COUNT = "SELECT_COUNT_";
public static final String UPDATE = "";//"UPDATE_";
public static final int UPDATE_OP = 2;
public Future<Integer> delete(String statement);
public Future<Integer> delete(String statement, Object parameter);
public Future<Integer> delete(String nameSpace, String statement, Object parameter);
public Future<Object> selectOne(String statement);
public Future<Object> selectOne(String statement, Object parameter);
public Future<Object> selectOne(String nameSpace, String statement, Object parameter);
public Future<List<?>> selectList(String statement);
public Future<List<?>> selectList(String statement, Object parameter);
public Future<List<?>> selectList(String nameSpace, String statement, Object parameter);
public Future<PageResult> selectPageList(String statement);
public Future<PageResult> selectPageList(String statement, Object parameter);
public Future<PageResult> selectPageList(String nameSpace, String statement, Object parameter);
public Future<Integer> selectCount(String statement);
public Future<Integer> selectCount(String statement, Object parameter);
public Future<Integer> selectCount(String nameSpace, String statement, Object parameter);
public Future<Integer> insert(String statement);
public Future<Integer> insert(String statement, Object parameter);
public Future<Integer> insert(String nameSpace, String statement, Object parameter);
public Future<Integer> update(String statement);
public Future<Integer> update(String statement, Object parameter);
public Future<Integer> update(String nameSpace, String statement, Object parameter);
}
... ...
/**
* Project: petstore-dal
*
* File Created at 2015年11月2日
* $Id$
*
* Copyright 2015 Yoho.cn Croporation Limited.
* All rights reserved.
*
* This software is the confidential and proprietary information of
* Yoho Company. ("Confidential Information"). You shall not
* disclose such Confidential Information and shall use it only in
* accordance with the terms of the license agreement you entered into
* with Yoho.cn.
*/
package com.taobao.sample.petstore.dal.dao;
import java.util.ArrayList;
import java.util.List;
/**
* TODO Comment of MainTest
*
* @author ibm
* @version $Id: MainTest.java 2015年11月2日 上午10:08:19 $
*
*/
public class MainTest {
/**
* @param args
*/
public static void main(String[] args) {
List list = new ArrayList();
for (int i = 0; i < 103; i++) {
list.add(i);
}
List sublist = new ArrayList();
for (int j=0; j < 103; j+=5 ) {
System.out.println(">>>>>>>>>>j:" + j);
int toindex = (j+5) > list.size()? list.size() : (j+5);
sublist = list.subList(j, toindex);
for (Object a : sublist) {
System.out.println(">>>>>>>>>>sublist:" + a.toString());
}
}
}
}
... ...
/**
* Project: petstore-dal
*
* File Created at 2015年11月1日
* $Id$
*
* Copyright 2015 Yoho.cn Croporation Limited.
* All rights reserved.
*
* This software is the confidential and proprietary information of
* Yoho Company. ("Confidential Information"). You shall not
* disclose such Confidential Information and shall use it only in
* accordance with the terms of the license agreement you entered into
* with Yoho.cn.
*/
package com.taobao.sample.petstore.dal.dao;
/**
* TODO Comment of Student
*
* @author ibm
* @version $Id: Student.java 2015年11月1日 上午2:26:07 $
*
*/
public class Student {
private String grade;
private String classs;
private String name;
public Student(){
}
public Student(String grade, String classs, String name){
this.grade = grade;
this.classs = classs;
this.name = name;
}
/**
* @return the grade
*/
public String getGrade() {
return grade;
}
/**
* @param grade the grade to set
*/
public void setGrade(String grade) {
this.grade = grade;
}
/**
* @return the classs
*/
public String getClasss() {
return classs;
}
/**
* @param classs the classs to set
*/
public void setClasss(String classs) {
this.classs = classs;
}
/**
* @return the name
*/
public String getName() {
return name;
}
/**
* @param name the name to set
*/
public void setName(String name) {
this.name = name;
}
}
... ...
/**
* Project: petstore-dal
*
* File Created at 2015年11月1日
* $Id$
*
* Copyright 2015 Yoho.cn Croporation Limited.
* All rights reserved.
*
* This software is the confidential and proprietary information of
* Yoho Company. ("Confidential Information"). You shall not
* disclose such Confidential Information and shall use it only in
* accordance with the terms of the license agreement you entered into
* with Yoho.cn.
*/
package com.taobao.sample.petstore.dal.dao;
import java.util.Iterator;
import java.util.Map;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.terms.StringTerms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms.Bucket;
import org.elasticsearch.search.aggregations.bucket.terms.TermsBuilder;
import org.elasticsearch.search.aggregations.bucket.terms.UnmappedTerms;
/**
* TODO Comment of TestAggregation
*
* @author ibm
* @version $Id: TestAggregation.java 2015年11月1日 上午2:35:13 $
*/
public class TestAggregation {
private int port = 9300;
private String ipAddress[] = { "127.0.0.1" };
private static Client client;
public TestAggregation() {
TransportClient transportClient = new TransportClient();
for (String ip : ipAddress) {
transportClient.addTransportAddress(new InetSocketTransportAddress(ip, port));
}
client = transportClient;
}
public void testAggregation() {
SearchRequestBuilder srb = new SearchRequestBuilder(client);
srb.setIndices("school");
srb.setTypes("student");
srb.setSearchType(SearchType.COUNT);
TermsBuilder gradeTermsBuilder = AggregationBuilders.terms("gradeAgg").field("grade");
TermsBuilder classTermsBuilder = AggregationBuilders.terms("classAgg").field("classs");
gradeTermsBuilder.subAggregation(classTermsBuilder);
srb.addAggregation(gradeTermsBuilder);
SearchResponse sr = srb.execute().actionGet();
Map<String, Aggregation> aggMap = sr.getAggregations().asMap();
StringTerms gradeTerms = (StringTerms) aggMap.get("gradeAgg");
Iterator<Bucket> gradeBucketIt = gradeTerms.getBuckets().iterator();
while (gradeBucketIt.hasNext()) {
Bucket gradeBucket = gradeBucketIt.next();
System.out.println(gradeBucket.getKey() + "年级有" + gradeBucket.getDocCount() + "个学生。");
StringTerms classTerms = (StringTerms) gradeBucket.getAggregations().asMap().get("classAgg");
Iterator<Bucket> classBucketIt = classTerms.getBuckets().iterator();
while (classBucketIt.hasNext()) {
Bucket classBucket = classBucketIt.next();
System.out.println(gradeBucket.getKey() + "年级" + classBucket.getKey() + "班有" + classBucket.getDocCount() + "个学生。");
}
System.out.println();
}
}
public static void main(String[] args) throws InterruptedException {
TestAggregation ta = new TestAggregation();
System.out.println(">>>>>>>>>>>>Start...");
Thread.sleep(30000);
ta.testAggregation();
Thread.sleep(30000);
System.out.println(">>>>>>>>>>>>");
ta.testAggregation();
client.close();
}
}
... ...
/**
* Project: petstore-dal
*
* File Created at 2015年10月29日
* $Id$
*
* Copyright 2011 Yoho.cn Croporation Limited.
* All rights reserved.
*
* This software is the confidential and proprietary information of
* Yoho Company. ("Confidential Information"). You shall not
* disclose such Confidential Information and shall use it only in
* accordance with the terms of the license agreement you entered into
* with Yoho.cn.
*/
package com.taobao.sample.petstore.dal.dao;
/**
* TODO Comment of User
*
* @author ibm
* @version $Id: User.java 2015年10月29日 下午5:05:35 $
*
*/
public class User {
private Long id;
private String name;
private Integer age;
public User(){
}
public User(Long id, String name, Integer age){
this.id = id;
this.name = name;
this.age = age;
}
/**
* @return the id
*/
public Long getId() {
return id;
}
/**
* @param id the id to set
*/
public void setId(Long id) {
this.id = id;
}
/**
* @return the name
*/
public String getName() {
return name;
}
/**
* @param name the name to set
*/
public void setName(String name) {
this.name = name;
}
/**
* @return the age
*/
public Integer getAge() {
return age;
}
/**
* @param age the age to set
*/
public void setAge(Integer age) {
this.age = age;
}
}
... ...
... ... @@ -10,6 +10,7 @@ import org.apache.commons.logging.LogFactory;
//import org.springframework.orm.ibatis.SqlMapClientCallback;
import org.mybatis.spring.support.SqlSessionDaoSupport;
import com.alibaba.fastjson.JSONObject;
import com.taobao.common.lang.diagnostic.Profiler;
import com.taobao.sample.petstore.dal.dao.GenericDao;
import com.taobao.sample.petstore.dal.dao.OperatorAdapter;
... ... @@ -28,52 +29,74 @@ import com.taobao.sample.petstore.dal.page.PageResult;
* @version $Id: BaseModel.java 2010-8-9 下午05:37:34 $
*/
public class IbatisGenericDaoImpl extends SqlSessionDaoSupport implements GenericDao {
static Log log = LogFactory.getLog(MybatisGenericDaoImpl.class);
public static final long DATABASE_OPERATE_TIMES_LIMITS = 500;
OperatorAdapter operatorAdapter;
protected static final Log log = LogFactory.getLog(MybatisGenericDaoImpl.class);
private long sqlTimeLimits = 500;
private boolean showSql = true;
OperatorAdapter operatorAdapter;
/**
* @return the sqlTimeLimits
*/
public long getSqlTimeLimits() {
return sqlTimeLimits;
}
/**
* @param sqlTimeLimits the sqlTimeLimits to set
*/
public void setSqlTimeLimits(long sqlTimeLimits) {
this.sqlTimeLimits = sqlTimeLimits;
}
/**
* @return the showSql
*/
public boolean isShowSql() {
return this.showSql;
}
/**
* @param showSql the showSql to set
*/
public void setShowSql(boolean showSql) {
this.showSql = showSql;
}
public void setOperatorAdapter(OperatorAdapter operatorAdapter) {
this.operatorAdapter = operatorAdapter;
}
/***********************************************************
* @param statement
***********************************************************/
public int delete(String statement) {
return delete(statement, null);
}
public int delete(String statement, Object parameter) {
isBaseDo(parameter);
return delete(null, statement, parameter);
}
public int delete(String nameSpace, String statement, Object parameter) {
Profiler.enter("delete(nameSpace=" + nameSpace + ",parameter=" + parameter + ",statement=" + statement + ")");
checkNameSpace(nameSpace);
setDefaultValues(parameter, DELETE_OP);
long startTime=System.currentTimeMillis(); //获取开始时间
long startTime = System.currentTimeMillis(); //获取开始时间
int i = 0;
statement = (nameSpace==null?"" : nameSpace + ".") + statement;
statement = (nameSpace == null ? "" : nameSpace + ".") + statement;
if (parameter == null) {
i = getSqlSession().delete(statement);
} else {
i = getSqlSession().delete(statement, parameter);
}
long endTime=System.currentTimeMillis(); //获取结束时间
if (log.isDebugEnabled()) {
long l = endTime - startTime;
if (l > DATABASE_OPERATE_TIMES_LIMITS) {
log.debug("[delete(nameSpace=" + nameSpace + ",parameter=" + parameter + ",statement=" + statement + ")]Delete object list from database in " + (endTime-startTime) + "ms");
}
}
profilerLog(startTime, statement, parameter); // 记录sql语句执行耗时
Profiler.release();
return i;
}
/**********************************************************************
* @param statement
**********************************************************************/
... ... @@ -89,23 +112,17 @@ public class IbatisGenericDaoImpl extends SqlSessionDaoSupport implements Generi
public Object selectOne(String nameSpace, String statement, Object parameter) {
Profiler.enter("selectOne(nameSpace=" + nameSpace + ",parameter=" + parameter + ",statement=" + statement + ")");
checkNameSpace(nameSpace);
long startTime=System.currentTimeMillis(); //获取开始时间
long startTime = System.currentTimeMillis(); //获取开始时间
Object o = null;
statement = (nameSpace==null?"" : nameSpace + ".") + statement;
statement = (nameSpace == null ? "" : nameSpace + ".") + statement;
if (parameter == null) {
o = getSqlSession().selectOne(statement);
} else {
o = getSqlSession().selectOne(statement, parameter);
}
long endTime=System.currentTimeMillis(); //获取结束时间
if (log.isDebugEnabled()) {
long l = endTime - startTime;
if (l > DATABASE_OPERATE_TIMES_LIMITS) {
log.debug("[getObj(nameSpace=" + nameSpace + ",parameter=" + parameter + ",statement=" + statement + ")]Get object from database in " + (endTime-startTime) + "ms");
}
}
profilerLog(startTime, statement, parameter); // 记录sql语句执行耗时
Profiler.release();
return o;
}
... ... @@ -113,56 +130,53 @@ public class IbatisGenericDaoImpl extends SqlSessionDaoSupport implements Generi
public List<?> selectList(String statement) {
return selectList(null, statement);
}
public List<?> selectList(String statement, Object parameter) {
isBaseDo(parameter);
return selectList(null, statement, parameter);
}
public List<?> selectList(String nameSpace, String statement, Object parameter) {
Profiler.enter("selectList(nameSpace=" + nameSpace + ",parameter=" + parameter + ",statement=" + statement + ")");
checkNameSpace(nameSpace);
long startTime=System.currentTimeMillis(); //获取开始时间
long startTime = System.currentTimeMillis(); //获取开始时间
List<?> list = null;
statement = (nameSpace==null?"" : nameSpace + ".") + statement;
statement = (nameSpace == null ? "" : nameSpace + ".") + statement;
if (parameter == null) {
list = getSqlSession().selectList(statement);
} else {
list = getSqlSession().selectList(statement, parameter);
}
long endTime=System.currentTimeMillis(); //获取结束时间
if (log.isDebugEnabled()) {
long l = endTime - startTime;
if (l > DATABASE_OPERATE_TIMES_LIMITS) {
log.debug("[getObj(nameSpace=" + nameSpace + ",parameter=" + parameter + ",statement=" + statement + ")]Get object from database in " + (endTime-startTime) + "ms");
}
}
profilerLog(startTime, statement, parameter); // 记录sql语句执行耗时
Profiler.release();
return list;
}
public PageResult selectPageList(String statement) {
return selectPageList(statement, null);
}
public PageResult selectPageList(String statement, Object parameter) {
isBaseDo(parameter);
return selectPageList(null, statement, parameter);
}
public PageResult selectPageList(String nameSpace, String statement, Object parameter) {
Profiler.enter("selectPageList(nameSpace=" + nameSpace + ",parameter=" + parameter + ",statement=" + statement + ")");
PageResult pageResult = new PageResult(); //构造一个分页器
if (parameter != null &&parameter instanceof BaseModel) {
if (parameter != null && parameter instanceof BaseModel) {
int total = selectCount(nameSpace, statement, parameter); //获取总记录数，以便计算分页数据
BaseModel baseModel = (BaseModel)parameter;
BaseModel baseModel = (BaseModel) parameter;
int page = baseModel.getCurrentPage(); //取得当前页
int pageSize = baseModel.getPageSize(); //取得每页记录数
pageResult.getPaginator().setItems(total);
pageResult.getPaginator().setItemsPerPage(pageSize);
pageResult.getPaginator().setPage(page + 1);
//获取分页数据计算结果
baseModel.setPageStart(pageResult.getPageStart());
baseModel.setPageStart(pageResult.getPageStart());
baseModel.setPageSize(pageResult.getPageSize());
}
pageResult.setResult(selectList(nameSpace, statement, parameter));
... ... @@ -173,13 +187,15 @@ public class IbatisGenericDaoImpl extends SqlSessionDaoSupport implements Generi
public int selectCount(String statement) {
return selectCount(statement, null);
}
public int selectCount(String statement, Object parameter) {
isBaseDo(parameter);
return selectCount(statement, parameter);
}
public int selectCount(String nameSpace, String statement, Object parameter) {
checkNameSpace(nameSpace);
statement = (nameSpace==null?"" : nameSpace + ".") + statement;
statement = (nameSpace == null ? "" : nameSpace + ".") + statement;
if (parameter == null) {
return (Integer) getSqlSession().selectOne(statement);
} else {
... ... @@ -187,7 +203,6 @@ public class IbatisGenericDaoImpl extends SqlSessionDaoSupport implements Generi
}
}
/**********************************************************************
* @param statement
**********************************************************************/
... ... @@ -204,29 +219,21 @@ public class IbatisGenericDaoImpl extends SqlSessionDaoSupport implements Generi
Profiler.enter("insert(nameSpace=" + nameSpace + ",parameter=" + parameter + ",statement=" + statement + ")");
checkNameSpace(nameSpace);
setDefaultValues(parameter, INSERT_OP);
long startTime=System.currentTimeMillis(); //获取开始时间
long startTime = System.currentTimeMillis(); //获取开始时间
Object o = null;
statement = (nameSpace == null? "" : nameSpace + ".") + statement;
statement = (nameSpace == null ? "" : nameSpace + ".") + statement;
if (parameter != null) {
o = getSqlSession().insert(statement, parameter);
} else {
o = getSqlSession().insert(statement);
}
long endTime=System.currentTimeMillis(); //获取结束时间
if (log.isDebugEnabled()) {
long l = endTime - startTime;
if (l > DATABASE_OPERATE_TIMES_LIMITS) {
log.debug("[insert(nameSpace=" + nameSpace + ",parameter=" + parameter + ",statement=" + statement + ")]Insert object into database in " + (endTime-startTime) + "ms");
}
}
profilerLog(startTime, statement, parameter); // 记录sql语句执行耗时
Profiler.release();
return o;
}
/**********************************************************************
* @param statement
**********************************************************************/
... ... @@ -243,28 +250,20 @@ public class IbatisGenericDaoImpl extends SqlSessionDaoSupport implements Generi
Profiler.enter("update(nameSpace=" + nameSpace + ",parameter=" + parameter + ",statement=" + statement + ")");
checkNameSpace(nameSpace);
setDefaultValues(parameter, UPDATE_OP);
long startTime=System.currentTimeMillis(); //获取开始时间
long startTime = System.currentTimeMillis(); //获取开始时间
int i = 0;
statement = (nameSpace==null?"" : nameSpace + ".") + statement;
statement = (nameSpace == null ? "" : nameSpace + ".") + statement;
if (parameter == null) {
i = getSqlSession().update(statement);
} else {
i = getSqlSession().update(statement, parameter);
}
long endTime=System.currentTimeMillis(); //获取结束时间
if (log.isDebugEnabled()) {
long l = endTime - startTime;
if (l > DATABASE_OPERATE_TIMES_LIMITS) {
log.debug("[update(nameSpace=" + nameSpace + ",parameter=" + parameter + ",statement=" + statement + ")]Update object from database in " + (endTime-startTime) + "ms");
}
}
profilerLog(startTime, statement, parameter); // 记录sql语句执行耗时
Profiler.release();
return i;
}
/**********************************************************************
* @param parameter
... ... @@ -284,44 +283,44 @@ public class IbatisGenericDaoImpl extends SqlSessionDaoSupport implements Generi
BaseModel data = (BaseModel) parameter;
if (data.isDefaultBiz()) {
switch (operation) {
case INSERT_OP:
data.setGmtCreate(now);
data.setGmtModified(now);
data.setIsDeleted(ModelConstant.NO);
if (operator != null) {
data.setCreator(operator);
data.setModifier(operator);
}
break;
case UPDATE_OP:
case DELETE_OP:
data.setGmtModified(new Date());
if (operator != null) {
data.setModifier(operator);
}
break;
case INSERT_OP:
data.setGmtCreate(now);
data.setGmtModified(now);
data.setIsDeleted(ModelConstant.NO);
if (operator != null) {
data.setCreator(operator);
data.setModifier(operator);
}
break;
case UPDATE_OP:
case DELETE_OP:
data.setGmtModified(new Date());
if (operator != null) {
data.setModifier(operator);
}
break;
}
}
} else if (parameter instanceof Map) {
Map data = (Map) parameter;
if (!data.containsKey(DEFAULT_BIZ) || Boolean.TRUE.equals(data.get(DEFAULT_BIZ))) {
switch (operation) {
case INSERT_OP:
data.put(ModelConstant.GMT_CREATE, now);
data.put(ModelConstant.GMT_MODIFIED, now);
if (operator != null) {
data.put(ModelConstant.CREATOR, operator);
data.put(ModelConstant.MODIFIER, operator);
}
data.put(ModelConstant.IS_DELETED, ModelConstant.NO);
break;
case UPDATE_OP:
case DELETE_OP:
data.put(ModelConstant.GMT_MODIFIED, now);
if (operator != null) {
((Map) parameter).put(ModelConstant.MODIFIER, operator);
}
break;
case INSERT_OP:
data.put(ModelConstant.GMT_CREATE, now);
data.put(ModelConstant.GMT_MODIFIED, now);
if (operator != null) {
data.put(ModelConstant.CREATOR, operator);
data.put(ModelConstant.MODIFIER, operator);
}
data.put(ModelConstant.IS_DELETED, ModelConstant.NO);
break;
case UPDATE_OP:
case DELETE_OP:
data.put(ModelConstant.GMT_MODIFIED, now);
if (operator != null) {
((Map) parameter).put(ModelConstant.MODIFIER, operator);
}
break;
}
}
}
... ... @@ -329,7 +328,6 @@ public class IbatisGenericDaoImpl extends SqlSessionDaoSupport implements Generi
log.error("Set default field failed!", e);
}
}
protected void isBaseDo(Object obj) {
if (!(obj instanceof BaseModel)) {
... ... @@ -342,7 +340,6 @@ public class IbatisGenericDaoImpl extends SqlSessionDaoSupport implements Generi
throw new RuntimeException("MUST give me a NameSpace for Ibtis SQL mapping.");
}
}
public static String toUpperCaseWithUnderscores(String str) {
StringBuffer result = new StringBuffer();
... ... @@ -361,4 +358,42 @@ public class IbatisGenericDaoImpl extends SqlSessionDaoSupport implements Generi
}
return result.toString();
}
/**
* 输出json串
*
* @param obj
* @return
*/
private String toJson(Object obj) {
return JSONObject.toJSONString(obj);
}
/**
* 获取当前执行的sql语句
*
* @param statement
* @param parameter
* @return
*/
private String getBoundSql(String statement, Object parameter) {
return getSqlSession().getConfiguration().getMappedStatement(statement).getBoundSql(parameter).getSql();
}
/**
* 记录超时的sql语句
*
* @param startTime
*/
private void profilerLog(long startTime, String statement, Object parameter) {
long endTime = System.currentTimeMillis(); //获取结束时间
long time = endTime - startTime;
if (time > sqlTimeLimits) {
if (this.isShowSql()) {
log.info("Excute sql [" + getBoundSql(statement, parameter) + " ;\r\n" + toJson(parameter) + "] in " + time + "ms");
} else {
log.info("Excute sql [statement=" + statement + ",parameter=" + toJson(parameter) + "] in " + time + "ms");
}
}
}
}
... ...
/**
* Project: petstore-dal
*
* File Created at 2015年10月29日
* $Id$
*
* Copyright 2011 Yoho.cn Croporation Limited.
* All rights reserved.
*
* This software is the confidential and proprietary information of
* Yoho Company. ("Confidential Information"). You shall not
* disclose such Confidential Information and shall use it only in
* accordance with the terms of the license agreement you entered into
* with Yoho.cn.
*/
package com.taobao.sample.petstore.dal.dao.mybatis;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.mybatis.spring.support.SqlSessionDaoSupport;
import com.alibaba.fastjson.JSONObject;
import com.taobao.common.lang.diagnostic.Profiler;
import com.taobao.sample.common.concurrent.ThreadHelper;
import com.taobao.sample.petstore.dal.dao.GenericAsyncDao;
import com.taobao.sample.petstore.dal.dao.OperatorAdapter;
import com.taobao.sample.petstore.dal.model.BaseModel;
import com.taobao.sample.petstore.dal.model.ModelConstant;
import com.taobao.sample.petstore.dal.page.PageResult;
/**
* 通用的Mybatis持久化操作异步调用实现类
*
* @author ibm
* @version $Id: MybatisAsyncDaoImpl.java 2015年10月29日 下午12:47:42 $
*/
public class MybatisAsyncDaoImpl extends SqlSessionDaoSupport implements GenericAsyncDao {
protected static final Log log = LogFactory.getLog(MybatisGenericDaoImpl.class);
private long sqlTimeLimits = 500;
private boolean showSql = true;
OperatorAdapter operatorAdapter;
ThreadHelper threadHelper;
/**
* @param threadHelper the threadHelper to set
*/
public void setThreadHelper(ThreadHelper threadHelper) {
this.threadHelper = threadHelper;
}
/**
* @return the sqlTimeLimits
*/
public long getSqlTimeLimits() {
return sqlTimeLimits;
}
/**
* @param sqlTimeLimits the sqlTimeLimits to set
*/
public void setSqlTimeLimits(long sqlTimeLimits) {
this.sqlTimeLimits = sqlTimeLimits;
}
/**
* @return the showSql
*/
public boolean isShowSql() {
return this.showSql;
}
/**
* @param showSql the showSql to set
*/
public void setShowSql(boolean showSql) {
this.showSql = showSql;
}
public void setOperatorAdapter(OperatorAdapter operatorAdapter) {
this.operatorAdapter = operatorAdapter;
}
/***********************************************************
* @param statement
***********************************************************/
public Future<Integer> delete(String statement) {
return delete(statement, null);
}
public Future<Integer> delete(String statement, Object parameter) {
isBaseDo(parameter);
return delete(null, statement, parameter);
}
public Future<Integer> delete(String nameSpace, String statement, Object parameter) {
Profiler.enter("delete(nameSpace=" + nameSpace + ",parameter=" + parameter + ",statement=" + statement + ")");
@SuppressWarnings("unchecked")
Future<Integer> future = (Future<Integer>) threadHelper.execute(new Callable<Integer>() {
@Override
public Integer call() throws Exception {
checkNameSpace(nameSpace);
setDefaultValues(parameter, DELETE_OP);
long startTime = System.currentTimeMillis(); //获取开始时间
Integer i = 0;
String sql = (nameSpace == null ? "" : nameSpace + ".") + statement;
if (parameter == null) {
i = getSqlSession().delete(sql);
} else {
i = getSqlSession().delete(sql, parameter);
}
profilerLog(startTime, sql, parameter); // 记录sql语句执行耗时
return i;
}
});
Profiler.release();
return future;
}
/**********************************************************************
* @param statement
**********************************************************************/
public Future<Object> selectOne(String statement) {
return selectOne(statement, null);
}
public Future<Object> selectOne(String statement, Object parameter) {
isBaseDo(parameter);
return selectOne(null, statement, parameter);
}
public Future<Object> selectOne(String nameSpace, String statement, Object parameter) {
Profiler.enter("selectOne(nameSpace=" + nameSpace + ",parameter=" + parameter + ",statement=" + statement + ")");
@SuppressWarnings("unchecked")
Future<Object> future = (Future<Object>) threadHelper.execute(new Callable<Object>() {
@Override
public Object call() throws Exception {
checkNameSpace(nameSpace);
long startTime = System.currentTimeMillis(); //获取开始时间
Object o = null;
String sql = (nameSpace == null ? "" : nameSpace + ".") + statement;
if (parameter == null) {
o = getSqlSession().selectOne(sql);
} else {
o = getSqlSession().selectOne(sql, parameter);
}
profilerLog(startTime, sql, parameter); // 记录sql语句执行耗时
return o;
}
});
Profiler.release();
return future;
}
public Future<List<?>> selectList(String statement) {
return selectList(null, statement);
}
public Future<List<?>> selectList(String statement, Object parameter) {
isBaseDo(parameter);
return selectList(null, statement, parameter);
}
public Future<List<?>> selectList(String nameSpace, String statement, Object parameter) {
Profiler.enter("selectList(nameSpace=" + nameSpace + ",parameter=" + parameter + ",statement=" + statement + ")");
@SuppressWarnings("unchecked")
Future<List<?>> future = (Future<List<?>>) threadHelper.execute(new Callable<List<?>>() {
@Override
public List<?> call() throws Exception {
checkNameSpace(nameSpace);
long startTime = System.currentTimeMillis(); //获取开始时间
List<?> list = null;
String sql = (nameSpace == null ? "" : nameSpace + ".") + statement;
if (parameter == null) {
list = getSqlSession().selectList(sql);
} else {
list = getSqlSession().selectList(sql, parameter);
}
profilerLog(startTime, sql, parameter); // 记录sql语句执行耗时
return list;
}
});
Profiler.release();
return future;
}
public Future<PageResult> selectPageList(String statement) {
return selectPageList(statement, null);
}
public Future<PageResult> selectPageList(String statement, Object parameter) {
isBaseDo(parameter);
return selectPageList(null, statement, parameter);
}
public Future<PageResult> selectPageList(String nameSpace, String statement, Object parameter) {
Profiler.enter("selectPageList(nameSpace=" + nameSpace + ",parameter=" + parameter + ",statement=" + statement + ")");
@SuppressWarnings("unchecked")
Future<PageResult> future = (Future<PageResult>) threadHelper.execute(new Callable<PageResult>() {
@SuppressWarnings("rawtypes")
@Override
public PageResult call() throws Exception {
PageResult pageResult = new PageResult(); //构造一个分页器
if (parameter != null && parameter instanceof BaseModel) {
int total = selectCount(nameSpace, statement, parameter).get(); //获取总记录数，以便计算分页数据
BaseModel baseModel = (BaseModel) parameter;
int page = baseModel.getCurrentPage(); //取得当前页
int pageSize = baseModel.getPageSize(); //取得每页记录数
pageResult.getPaginator().setItems(total);
pageResult.getPaginator().setItemsPerPage(pageSize);
pageResult.getPaginator().setPage(page + 1);
//获取分页数据计算结果
baseModel.setPageStart(pageResult.getPageStart());
baseModel.setPageSize(pageResult.getPageSize());
}
Future<List<Object>> ft = (Future) selectList(nameSpace, statement, parameter);
pageResult.setResult(ft.get());
return pageResult;
}
});
Profiler.release();
return future;
}
public Future<Integer> selectCount(String statement) {
return selectCount(statement, null);
}
public Future<Integer> selectCount(String statement, Object parameter) {
isBaseDo(parameter);
return selectCount(statement, parameter);
}
public Future<Integer> selectCount(String nameSpace, String statement, Object parameter) {
@SuppressWarnings("unchecked")
Future<Integer> future = (Future<Integer>) threadHelper.execute(new Callable<Integer>() {
@Override
public Integer call() throws Exception {
checkNameSpace(nameSpace);
String sql = (nameSpace == null ? "" : nameSpace + ".") + statement;
if (parameter == null) {
return (Integer) getSqlSession().selectOne(sql);
} else {
return (Integer) getSqlSession().selectOne(sql, parameter);
}
}
});
return future;
}
/**********************************************************************
* @param statement
**********************************************************************/
public Future<Integer> insert(String statement) {
return insert(statement, null);
}
public Future<Integer> insert(String statement, Object parameter) {
isBaseDo(parameter);
return insert(null, statement, parameter);
}
public Future<Integer> insert(String nameSpace, String statement, Object parameter) {
Profiler.enter("insert(nameSpace=" + nameSpace + ",parameter=" + parameter + ",statement=" + statement + ")");
@SuppressWarnings("unchecked")
Future<Integer> future = (Future<Integer>) threadHelper.execute(new Callable<Integer>() {
@Override
public Integer call() throws Exception {
checkNameSpace(nameSpace);
setDefaultValues(parameter, INSERT_OP);
long startTime = System.currentTimeMillis(); //获取开始时间
int i = 0;
String sql = (nameSpace == null ? "" : nameSpace + ".") + statement;
if (parameter != null) {
i = getSqlSession().insert(sql, parameter);
} else {
i = getSqlSession().insert(sql);
}
profilerLog(startTime, sql, parameter); // 记录sql语句执行耗时
return i;
}
});
Profiler.release();
return future;
}
/**********************************************************************
* @param statement
**********************************************************************/
public Future<Integer> update(String statement) {
return update(statement, null);
}
public Future<Integer> update(String statement, Object parameter) {
isBaseDo(parameter);
return update(null, statement, parameter);
}
public Future<Integer> update(String nameSpace, String statement, Object parameter) {
Profiler.enter("update(nameSpace=" + nameSpace + ",parameter=" + parameter + ",statement=" + statement + ")");
@SuppressWarnings("unchecked")
Future<Integer> future = (Future<Integer>) threadHelper.execute(new Callable<Integer>() {
@Override
public Integer call() throws Exception {
checkNameSpace(nameSpace);
setDefaultValues(parameter, UPDATE_OP);
long startTime = System.currentTimeMillis(); //获取开始时间
int i = 0;
String sql = (nameSpace == null ? "" : nameSpace + ".") + statement;
if (parameter == null) {
i = getSqlSession().update(sql);
} else {
i = getSqlSession().update(sql, parameter);
}
profilerLog(startTime, sql, parameter); // 记录sql语句执行耗时
return i;
}
});
Profiler.release();
return future;
}
/**********************************************************************
* @param parameter
* @param operation
**********************************************************************/
@SuppressWarnings({ "unchecked", "rawtypes" })
protected void setDefaultValues(Object parameter, int operation) {
String operator = null;
if (operatorAdapter == null) {
log.info("No operator information configured in DAO.");
} else {
operator = operatorAdapter.getOperator();
}
try {
Date now = new java.util.Date();
if (parameter instanceof BaseModel) {
BaseModel data = (BaseModel) parameter;
if (data.isDefaultBiz()) {
switch (operation) {
case INSERT_OP:
data.setGmtCreate(now);
data.setGmtModified(now);
data.setIsDeleted(ModelConstant.NO);
if (operator != null) {
data.setCreator(operator);
data.setModifier(operator);
}
break;
case UPDATE_OP:
case DELETE_OP:
data.setGmtModified(new Date());
if (operator != null) {
data.setModifier(operator);
}
break;
}
}
} else if (parameter instanceof Map) {
Map data = (Map) parameter;
if (!data.containsKey(DEFAULT_BIZ) || Boolean.TRUE.equals(data.get(DEFAULT_BIZ))) {
switch (operation) {
case INSERT_OP:
data.put(ModelConstant.GMT_CREATE, now);
data.put(ModelConstant.GMT_MODIFIED, now);
if (operator != null) {
data.put(ModelConstant.CREATOR, operator);
data.put(ModelConstant.MODIFIER, operator);
}
data.put(ModelConstant.IS_DELETED, ModelConstant.NO);
break;
case UPDATE_OP:
case DELETE_OP:
data.put(ModelConstant.GMT_MODIFIED, now);
if (operator != null) {
((Map) parameter).put(ModelConstant.MODIFIER, operator);
}
break;
}
}
}
} catch (Exception e) {
log.error("Set default field failed!", e);
}
}
protected void isBaseDo(Object obj) {
if (!(obj instanceof BaseModel)) {
throw new RuntimeException("BaseDo expected! Or you MUST give me a NameSpace for Ibtis SQL mapping.");
}
}
protected void checkNameSpace(String nameSpace) {
if (nameSpace == null) {
throw new RuntimeException("MUST give me a NameSpace for Ibtis SQL mapping.");
}
}
public static String toUpperCaseWithUnderscores(String str) {
StringBuffer result = new StringBuffer();
if (str != null) {
boolean lastIsNum = false;
result.append(str.charAt(0));
for (int i = 1; i < str.length(); i++) {
char aChar = str.charAt(i);
boolean thisIsNum = (aChar >= '0' && aChar <= '9');
if ((aChar >= 'A' && aChar <= 'Z') || (lastIsNum != thisIsNum)) {
result.append('_');
}
lastIsNum = thisIsNum;
result.append(Character.toUpperCase(aChar));
}
}
return result.toString();
}
/**
* 输出json串
*
* @param obj
* @return
*/
private String toJson(Object obj) {
return JSONObject.toJSONString(obj);
}
/**
* 获取当前执行的sql语句
*
* @param statement
* @param parameter
* @return
*/
private String getBoundSql(String statement, Object parameter) {
return getSqlSession().getConfiguration().getMappedStatement(statement).getBoundSql(parameter).getSql();
}
/**
* 记录超时的sql语句
*
* @param startTime
*/
private void profilerLog(long startTime, String statement, Object parameter) {
long endTime = System.currentTimeMillis(); //获取结束时间
long time = endTime - startTime;
if (time > sqlTimeLimits) {
if (this.isShowSql()) {
log.info("Excute sql [" + getBoundSql(statement, parameter) + " ;\r\n" + toJson(parameter) + "] in " + time + "ms");
} else {
log.info("Excute sql [statement=" + statement + ",parameter=" + toJson(parameter) + "] in " + time + "ms");
}
}
}
}
... ...
... ... @@ -8,6 +8,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.mybatis.spring.support.SqlSessionDaoSupport;
import com.alibaba.fastjson.JSONObject;
import com.taobao.common.lang.diagnostic.Profiler;
import com.taobao.sample.petstore.dal.dao.GenericDao;
import com.taobao.sample.petstore.dal.dao.OperatorAdapter;
... ... @@ -16,58 +17,80 @@ import com.taobao.sample.petstore.dal.model.ModelConstant;
import com.taobao.sample.petstore.dal.page.PageResult;
/**
* TODO Comment of BaseModel
* 通用的Mybatis持久化操作实现类
*
* @author 李健
* @version $Id: BaseModel.java 2010-8-9 下午05:37:34 $
*/
public class MybatisGenericDaoImpl extends SqlSessionDaoSupport implements GenericDao {
static Log log = LogFactory.getLog(MybatisGenericDaoImpl.class);
public static final long DATABASE_OPERATE_TIMES_LIMITS = 500;
OperatorAdapter operatorAdapter;
protected static final Log log = LogFactory.getLog(MybatisGenericDaoImpl.class);
private long sqlTimeLimits = 500;
private boolean showSql = true;
OperatorAdapter operatorAdapter;
/**
* @return the sqlTimeLimits
*/
public long getSqlTimeLimits() {
return sqlTimeLimits;
}
/**
* @param sqlTimeLimits the sqlTimeLimits to set
*/
public void setSqlTimeLimits(long sqlTimeLimits) {
this.sqlTimeLimits = sqlTimeLimits;
}
/**
* @return the showSql
*/
public boolean isShowSql() {
return this.showSql;
}
/**
* @param showSql the showSql to set
*/
public void setShowSql(boolean showSql) {
this.showSql = showSql;
}
public void setOperatorAdapter(OperatorAdapter operatorAdapter) {
this.operatorAdapter = operatorAdapter;
}
/***********************************************************
* @param statement
***********************************************************/
public int delete(String statement) {
return delete(statement, null);
}
public int delete(String statement, Object parameter) {
isBaseDo(parameter);
return delete(null, statement, parameter);
}
public int delete(String nameSpace, String statement, Object parameter) {
Profiler.enter("delete(nameSpace=" + nameSpace + ",parameter=" + parameter + ",statement=" + statement + ")");
checkNameSpace(nameSpace);
setDefaultValues(parameter, DELETE_OP);
long startTime=System.currentTimeMillis(); //获取开始时间
long startTime = System.currentTimeMillis(); //获取开始时间
int i = 0;
statement = (nameSpace==null?"" : nameSpace + ".") + statement;
statement = (nameSpace == null ? "" : nameSpace + ".") + statement;
if (parameter == null) {
i = getSqlSession().delete(statement);
} else {
i = getSqlSession().delete(statement, parameter);
}
long endTime=System.currentTimeMillis(); //获取结束时间
if (log.isDebugEnabled()) {
long l = endTime - startTime;
if (l > DATABASE_OPERATE_TIMES_LIMITS) {
log.debug("[delete(nameSpace=" + nameSpace + ",parameter=" + parameter + ",statement=" + statement + ")]Delete object list from database in " + (endTime-startTime) + "ms");
}
}
profilerLog(startTime, statement, parameter); // 记录sql语句执行耗时
Profiler.release();
return i;
}
/**********************************************************************
* @param statement
**********************************************************************/
... ... @@ -83,23 +106,17 @@ public class MybatisGenericDaoImpl extends SqlSessionDaoSupport implements Gener
public Object selectOne(String nameSpace, String statement, Object parameter) {
Profiler.enter("selectOne(nameSpace=" + nameSpace + ",parameter=" + parameter + ",statement=" + statement + ")");
checkNameSpace(nameSpace);
long startTime=System.currentTimeMillis(); //获取开始时间
long startTime = System.currentTimeMillis(); //获取开始时间
Object o = null;
statement = (nameSpace==null?"" : nameSpace + ".") + statement;
statement = (nameSpace == null ? "" : nameSpace + ".") + statement;
if (parameter == null) {
o = getSqlSession().selectOne(statement);
} else {
o = getSqlSession().selectOne(statement, parameter);
}
long endTime=System.currentTimeMillis(); //获取结束时间
if (log.isDebugEnabled()) {
long l = endTime - startTime;
if (l > DATABASE_OPERATE_TIMES_LIMITS) {
log.debug("[getObj(nameSpace=" + nameSpace + ",parameter=" + parameter + ",statement=" + statement + ")]Get object from database in " + (endTime-startTime) + "ms");
}
}
profilerLog(startTime, statement, parameter); // 记录sql语句执行耗时
Profiler.release();
return o;
}
... ... @@ -107,56 +124,53 @@ public class MybatisGenericDaoImpl extends SqlSessionDaoSupport implements Gener
public List<?> selectList(String statement) {
return selectList(null, statement);
}
public List<?> selectList(String statement, Object parameter) {
isBaseDo(parameter);
return selectList(null, statement, parameter);
}
public List<?> selectList(String nameSpace, String statement, Object parameter) {
Profiler.enter("selectList(nameSpace=" + nameSpace + ",parameter=" + parameter + ",statement=" + statement + ")");
checkNameSpace(nameSpace);
long startTime=System.currentTimeMillis(); //获取开始时间
long startTime = System.currentTimeMillis(); //获取开始时间
List<?> list = null;
statement = (nameSpace==null?"" : nameSpace + ".") + statement;
statement = (nameSpace == null ? "" : nameSpace + ".") + statement;
if (parameter == null) {
list = getSqlSession().selectList(statement);
} else {
list = getSqlSession().selectList(statement, parameter);
}
long endTime=System.currentTimeMillis(); //获取结束时间
if (log.isDebugEnabled()) {
long l = endTime - startTime;
if (l > DATABASE_OPERATE_TIMES_LIMITS) {
log.debug("[getObj(nameSpace=" + nameSpace + ",parameter=" + parameter + ",statement=" + statement + ")]Get object from database in " + (endTime-startTime) + "ms");
}
}
profilerLog(startTime, statement, parameter); // 记录sql语句执行耗时
Profiler.release();
return list;
}
public PageResult selectPageList(String statement) {
return selectPageList(statement, null);
}
public PageResult selectPageList(String statement, Object parameter) {
isBaseDo(parameter);
return selectPageList(null, statement, parameter);
}
public PageResult selectPageList(String nameSpace, String statement, Object parameter) {
Profiler.enter("selectPageList(nameSpace=" + nameSpace + ",parameter=" + parameter + ",statement=" + statement + ")");
PageResult pageResult = new PageResult(); //构造一个分页器
if (parameter != null &&parameter instanceof BaseModel) {
if (parameter != null && parameter instanceof BaseModel) {
int total = selectCount(nameSpace, statement, parameter); //获取总记录数，以便计算分页数据
BaseModel baseModel = (BaseModel)parameter;
BaseModel baseModel = (BaseModel) parameter;
int page = baseModel.getCurrentPage(); //取得当前页
int pageSize = baseModel.getPageSize(); //取得每页记录数
pageResult.getPaginator().setItems(total);
pageResult.getPaginator().setItemsPerPage(pageSize);
pageResult.getPaginator().setPage(page + 1);
//获取分页数据计算结果
baseModel.setPageStart(pageResult.getPageStart());
baseModel.setPageStart(pageResult.getPageStart());
baseModel.setPageSize(pageResult.getPageSize());
}
pageResult.setResult(selectList(nameSpace, statement, parameter));
... ... @@ -167,13 +181,15 @@ public class MybatisGenericDaoImpl extends SqlSessionDaoSupport implements Gener
public int selectCount(String statement) {
return selectCount(statement, null);
}
public int selectCount(String statement, Object parameter) {
isBaseDo(parameter);
return selectCount(statement, parameter);
}
public int selectCount(String nameSpace, String statement, Object parameter) {
checkNameSpace(nameSpace);
statement = (nameSpace==null?"" : nameSpace + ".") + statement;
statement = (nameSpace == null ? "" : nameSpace + ".") + statement;
if (parameter == null) {
return (Integer) getSqlSession().selectOne(statement);
} else {
... ... @@ -181,7 +197,6 @@ public class MybatisGenericDaoImpl extends SqlSessionDaoSupport implements Gener
}
}
/**********************************************************************
* @param statement
**********************************************************************/
... ... @@ -198,29 +213,21 @@ public class MybatisGenericDaoImpl extends SqlSessionDaoSupport implements Gener
Profiler.enter("insert(nameSpace=" + nameSpace + ",parameter=" + parameter + ",statement=" + statement + ")");
checkNameSpace(nameSpace);
setDefaultValues(parameter, INSERT_OP);
long startTime=System.currentTimeMillis(); //获取开始时间
long startTime = System.currentTimeMillis(); //获取开始时间
Object o = null;
statement = (nameSpace == null? "" : nameSpace + ".") + statement;
statement = (nameSpace == null ? "" : nameSpace + ".") + statement;
if (parameter != null) {
o = getSqlSession().insert(statement, parameter);
} else {
o = getSqlSession().insert(statement);
}
long endTime=System.currentTimeMillis(); //获取结束时间
if (log.isDebugEnabled()) {
long l = endTime - startTime;
if (l > DATABASE_OPERATE_TIMES_LIMITS) {
log.debug("[insert(nameSpace=" + nameSpace + ",parameter=" + parameter + ",statement=" + statement + ")]Insert object into database in " + (endTime-startTime) + "ms");
}
}
profilerLog(startTime, statement, parameter); // 记录sql语句执行耗时
Profiler.release();
return o;
}
/**********************************************************************
* @param statement
**********************************************************************/
... ... @@ -237,28 +244,20 @@ public class MybatisGenericDaoImpl extends SqlSessionDaoSupport implements Gener
Profiler.enter("update(nameSpace=" + nameSpace + ",parameter=" + parameter + ",statement=" + statement + ")");
checkNameSpace(nameSpace);
setDefaultValues(parameter, UPDATE_OP);
long startTime=System.currentTimeMillis(); //获取开始时间
long startTime = System.currentTimeMillis(); //获取开始时间
int i = 0;
statement = (nameSpace==null?"" : nameSpace + ".") + statement;
statement = (nameSpace == null ? "" : nameSpace + ".") + statement;
if (parameter == null) {
i = getSqlSession().update(statement);
} else {
i = getSqlSession().update(statement, parameter);
}
long endTime=System.currentTimeMillis(); //获取结束时间
if (log.isDebugEnabled()) {
long l = endTime - startTime;
if (l > DATABASE_OPERATE_TIMES_LIMITS) {
log.debug("[update(nameSpace=" + nameSpace + ",parameter=" + parameter + ",statement=" + statement + ")]Update object from database in " + (endTime-startTime) + "ms");
}
}
profilerLog(startTime, statement, parameter); // 记录sql语句执行耗时
Profiler.release();
return i;
}
/**********************************************************************
* @param parameter
... ... @@ -278,44 +277,44 @@ public class MybatisGenericDaoImpl extends SqlSessionDaoSupport implements Gener
BaseModel data = (BaseModel) parameter;
if (data.isDefaultBiz()) {
switch (operation) {
case INSERT_OP:
data.setGmtCreate(now);
data.setGmtModified(now);
data.setIsDeleted(ModelConstant.NO);
if (operator != null) {
data.setCreator(operator);
data.setModifier(operator);
}
break;
case UPDATE_OP:
case DELETE_OP:
data.setGmtModified(new Date());
if (operator != null) {
data.setModifier(operator);
}
break;
case INSERT_OP:
data.setGmtCreate(now);
data.setGmtModified(now);
data.setIsDeleted(ModelConstant.NO);
if (operator != null) {
data.setCreator(operator);
data.setModifier(operator);
}
break;
case UPDATE_OP:
case DELETE_OP:
data.setGmtModified(new Date());
if (operator != null) {
data.setModifier(operator);
}
break;
}
}
} else if (parameter instanceof Map) {
Map data = (Map) parameter;
if (!data.containsKey(DEFAULT_BIZ) || Boolean.TRUE.equals(data.get(DEFAULT_BIZ))) {
switch (operation) {
case INSERT_OP:
data.put(ModelConstant.GMT_CREATE, now);
data.put(ModelConstant.GMT_MODIFIED, now);
if (operator != null) {
data.put(ModelConstant.CREATOR, operator);
data.put(ModelConstant.MODIFIER, operator);
}
data.put(ModelConstant.IS_DELETED, ModelConstant.NO);
break;
case UPDATE_OP:
case DELETE_OP:
data.put(ModelConstant.GMT_MODIFIED, now);
if (operator != null) {
((Map) parameter).put(ModelConstant.MODIFIER, operator);
}
break;
case INSERT_OP:
data.put(ModelConstant.GMT_CREATE, now);
data.put(ModelConstant.GMT_MODIFIED, now);
if (operator != null) {
data.put(ModelConstant.CREATOR, operator);
data.put(ModelConstant.MODIFIER, operator);
}
data.put(ModelConstant.IS_DELETED, ModelConstant.NO);
break;
case UPDATE_OP:
case DELETE_OP:
data.put(ModelConstant.GMT_MODIFIED, now);
if (operator != null) {
((Map) parameter).put(ModelConstant.MODIFIER, operator);
}
break;
}
}
}
... ... @@ -323,7 +322,6 @@ public class MybatisGenericDaoImpl extends SqlSessionDaoSupport implements Gener
log.error("Set default field failed!", e);
}
}
protected void isBaseDo(Object obj) {
if (!(obj instanceof BaseModel)) {
... ... @@ -336,7 +334,6 @@ public class MybatisGenericDaoImpl extends SqlSessionDaoSupport implements Gener
throw new RuntimeException("MUST give me a NameSpace for Ibtis SQL mapping.");
}
}
public static String toUpperCaseWithUnderscores(String str) {
StringBuffer result = new StringBuffer();
... ... @@ -355,4 +352,42 @@ public class MybatisGenericDaoImpl extends SqlSessionDaoSupport implements Gener
}
return result.toString();
}
/**
* 输出json串
*
* @param obj
* @return
*/
private String toJson(Object obj) {
return JSONObject.toJSONString(obj);
}
/**
* 获取当前执行的sql语句
*
* @param statement
* @param parameter
* @return
*/
private String getBoundSql(String statement, Object parameter) {
return getSqlSession().getConfiguration().getMappedStatement(statement).getBoundSql(parameter).getSql();
}
/**
* 记录超时的sql语句
*
* @param startTime
*/
private void profilerLog(long startTime, String statement, Object parameter) {
long endTime = System.currentTimeMillis(); //获取结束时间
long time = endTime - startTime;
if (time > sqlTimeLimits) {
if (this.isShowSql()) {
log.info("Excute sql [" + getBoundSql(statement, parameter) + " ;\r\n" + toJson(parameter) + "] in " + time + "ms");
} else {
log.info("Excute sql [statement=" + statement + ",parameter=" + toJson(parameter) + "] in " + time + "ms");
}
}
}
}
... ...
package com.taobao.sample.petstore.dal.elasticsearch.callback;
import org.elasticsearch.action.ActionFuture;
import org.elasticsearch.action.ActionResponse;
import org.elasticsearch.client.Client;
/**
* 索引器操作的回调
*
* @param <T>
*/
public interface ClientCallback<T extends ActionResponse> {
/**
* 通过客户端执行Action
* @param client
* @return
*/
ActionFuture<T> execute(final Client client);
}
... ...
package com.taobao.sample.petstore.dal.elasticsearch.callback;
import org.elasticsearch.action.ActionFuture;
import org.elasticsearch.action.ActionResponse;
import org.elasticsearch.client.ClusterAdminClient;
/**
* 集群操作回调
*
* @param <T>
*/
public interface ClusterCallback<T extends ActionResponse> {
/**
*
* @param admin
* @return
*/
ActionFuture<T> execute(ClusterAdminClient admin);
}
... ...
package com.taobao.sample.petstore.dal.elasticsearch.callback;
import org.elasticsearch.action.ActionFuture;
import org.elasticsearch.action.ActionResponse;
import org.elasticsearch.client.IndicesAdminClient;
/**
* 节点操作回调
*
* @param <T>
*/
public interface NodeCallback<T extends ActionResponse> {
/**
* 接口索引操作客户端
* @param client
* @return
*/
ActionFuture<T> execute(final IndicesAdminClient client);
}
... ...
/**
* Project: petstore-dal
*
* File Created at 2015年11月4日
* $Id$
*
* Copyright 2015 Yoho.cn Croporation Limited.
* All rights reserved.
*
* This software is the confidential and proprietary information of
* Yoho Company. ("Confidential Information"). You shall not
* disclose such Confidential Information and shall use it only in
* accordance with the terms of the license agreement you entered into
* with Yoho.cn.
*/
package com.taobao.sample.petstore.dal.elasticsearch.client;
import java.util.List;
import java.util.Map;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.deletebyquery.DeleteByQueryResponse;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.update.UpdateResponse;
/**
* 调用Elasticsearch的模板类
*
* 该类主要用户搜索管理端，提供各种搜索查询、索引更新管理等操作
*
* 搜索服务端请使用<class>ElasticClientTemplate</class>
*
* @author jack.lee
* @version $Id: ElasticAdminClient.java 2015年11月4日 上午2:07:17 $
*
*/
public interface ElasticAdminClientTemplate extends ElasticClientTemplate {
/**
* 判断索引是否存在
*
* @param indexName 索引名称
* @return
*/
public boolean indexExists(final String indexName);
/**
* 判断索引是否健康
*
* @param indexName 索引名称
* @return
*/
public boolean indexHealth(final String indexName);
/**
* 创建索引
*
* @param indexName 索引名称
* @param properties 索引配置
* @param force 是否强制创建索引，不论索引是否已经存在
*/
public void createIndex(final String indexName, final Map<String, String> properties, final String mappingContent, boolean force);
/**
* 创建索引
*
* @param indexName 索引名称
* @param tempIndexName 临时索引名称
* @param properties 索引配置
* @param force 是否强制创建索引，不论索引是否已经存在
*/
public void createIndex(final String indexName, final String tempIndexName, final Map<String, String> properties, final String mappingContent, boolean force);
/**
* 删除索引
*
* @param indexName 索引名称
*/
public void deleteIndex(final String indexName);
/**
* 更新索引缓存
*
* @param indexName 索引名称
*/
public void refreshIndex(final String indexName);
/**
* 关闭索引
*
* @param indexName 索引名称
*/
public void closeIndex(final String indexName);
/**
* 刷新索引文件
*
* @param indexName 索引名称
*/
public void flushIndex(final String indexName);
/**
* 索引替换:将老索引的索引名加到新索引上，并删除老索引
*
* @param oldIndexName 旧索引名
* @param newIndexName 新索引名
*/
public void replaceIndex(final String oldIndexName, final String newIndexName);
/**
* 删除因系统停机或报错而产生未被删除的临时索引
* @param indexName
*/
public void deleteTempIndex(final String indexName);
/**
* 执行索引优化
*/
public void optimize(final String indexName);
/**
* 添加索引数据
*
* @param indexName 索引名称
* @param id 索引ID
* @param data 索引数据
*/
public IndexResponse addIndexData(final String indexName, final String id, final Object data);
/**
* 添加索引数据
*
* @param indexName 索引名称
* @param datas 索引数据列表
*/
public BulkResponse addIndexData(final String indexName, final List<?> datas);
public BulkResponse addIndexDataBean(final String indexName, final List<?> datas);
/**
* 删除索引数据
*
* @param indexName
*/
public DeleteResponse deleteIndexData(final String indexName, String id);
/**
* 根据query删除索引数据
* @param indexName
* @param query
*/
public DeleteByQueryResponse deleteIndexDataByQuery(final String indexName, String query);
/**
* 更新索引数据
*
* @param indexName
*/
public UpdateResponse updateIndexData(final String indexName, final String id, final Object data);
public UpdateResponse updateIndexDataStr(final String indexName, final String id, final String jsonData);
/**
* 批量更新（删除）数据
* @param indexName
* @param datas 待更新的数据列表
*/
public BulkResponse bulkUpdateIndexData(final String indexName, @SuppressWarnings("rawtypes") final List<Map> datas);
/**
* 获取所有相关的索引
* @param name
* @return
*/
public List<String> getRelatedIndexs(String name);
/**
* 索引配置
* @param indexName
* @param setting
*/
public void setIndexSettings(String indexName, Map<String, String> setting);
/**
* 在批量插入数据之前，进行索引优化配置
* 返回优化配置之前的配置
* @param indexName
* @return
*/
public Map<String, String> optimumSettingsForBulkIndexing(String indexName);
}
... ...
/**
* Project: petstore-dal
*
* File Created at 2015年11月4日
* $Id$
*
* Copyright 2015 Yoho.cn Croporation Limited.
* All rights reserved.
*
* This software is the confidential and proprietary information of
* Yoho Company. ("Confidential Information"). You shall not
* disclose such Confidential Information and shall use it only in
* accordance with the terms of the license agreement you entered into
* with Yoho.cn.
*/
package com.taobao.sample.petstore.dal.elasticsearch.client;
import java.util.List;
import java.util.Map;
import java.util.Set;
import com.taobao.sample.petstore.dal.elasticsearch.entity.ElasticQuery;
import com.taobao.sample.petstore.dal.elasticsearch.entity.ElasticResult;
/**
* 调用Elasticsearch的模板类
*
* 该类主要用户搜索服务端，提供各种搜索查询操作
*
* 搜索管理端请使用<class>ElasticAdminClientTemplate</class>
*
* @author jack.lee
* @version $Id: ElasticsearchClient.java 2015年11月4日 上午1:23:04 $
*
*/
public interface ElasticClientTemplate {
/**
* 获取索引器名称
*
* @return
*/
public String getName();
/**
* 设置索引器名称
*
* @param name
*/
public void setName(String name);
/**
* 通过关键词搜索数据
* @param indexName
* @param ElasticQuery
* @return
*/
public ElasticResult search(final String[] indexName, final ElasticQuery ElasticQuery);
/**
* 通过关键词搜索数据
* @param indexName
* @param ElasticQuery
* @return
*/
public ElasticResult search(final String indexName, final ElasticQuery ElasticQuery);
/**
* 通过id检索数据
*
* @param indexName
* @param ElasticQuery
* @return
*/
public List<Map<String, Object>> multiGet(final String indexName, final Set<String> idList, final List<String> fields);
/**
* 通过id检索数据
*
* @param indexName
* @param ElasticQuery
* @return
*/
public List<Map<String, Object>> multiGet(final String indexName, final String[] idList, final List<String> fields);
/**
* 通过id检索数据
*
* @param indexName
* @param ElasticQuery
* @return
*/
public Map<String, Object> get(final String indexName, final String id);
}
... ...
/**
* Project: petstore-dal
*
* File Created at 2015年11月4日
* $Id$
*
* Copyright 2015 Yoho.cn Croporation Limited.
* All rights reserved.
*
* This software is the confidential and proprietary information of
* Yoho Company. ("Confidential Information"). You shall not
* disclose such Confidential Information and shall use it only in
* accordance with the terms of the license agreement you entered into
* with Yoho.cn.
*/
package com.taobao.sample.petstore.dal.elasticsearch.client.impl;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.regex.Pattern;
import org.apache.commons.lang.time.DateFormatUtils;
import org.elasticsearch.action.ActionFuture;
import org.elasticsearch.action.ActionResponse;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthStatus;
import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest;
import org.elasticsearch.action.admin.indices.close.CloseIndexResponse;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexResponse;
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse;
import org.elasticsearch.action.admin.indices.flush.FlushResponse;
import org.elasticsearch.action.admin.indices.optimize.OptimizeResponse;
import org.elasticsearch.action.admin.indices.refresh.RefreshResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.deletebyquery.DeleteByQueryRequest;
import org.elasticsearch.action.deletebyquery.DeleteByQueryResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.update.UpdateRequestBuilder;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.ClusterAdminClient;
import org.elasticsearch.client.IndicesAdminClient;
import org.elasticsearch.client.Requests;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.metadata.AliasMetaData;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.cluster.metadata.MetaData;
import org.elasticsearch.common.settings.ImmutableSettings;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.taobao.sample.petstore.dal.elasticsearch.callback.ClientCallback;
import com.taobao.sample.petstore.dal.elasticsearch.callback.ClusterCallback;
import com.taobao.sample.petstore.dal.elasticsearch.callback.NodeCallback;
import com.taobao.sample.petstore.dal.elasticsearch.client.ElasticAdminClientTemplate;
/**
* TODO Comment of ElasticAdminClientImpl
*
* @author ibm
* @version $Id: ElasticAdminClientImpl.java 2015年11月4日 上午2:09:11 $
*/
public class ElasticAdminClientTemplateImpl extends ElasticClientTemplateImpl implements ElasticAdminClientTemplate {
public ElasticAdminClientTemplateImpl(Client client) {
super(client);
}
/**
* 判断索引是否存在
*
* @param indexName 索引名
* @return
*/
@Override
public boolean indexExists(String indexName) {
boolean exists = false;
final String indexId = this.getIndexId(indexName);
if (indexId != null) {
// 等待服务集群就绪
executeGet(new ClusterCallback<ClusterHealthResponse>() {
@Override
public ActionFuture<ClusterHealthResponse> execute(final ClusterAdminClient admin) {
return admin.health(Requests.clusterHealthRequest().waitForStatus(ClusterHealthStatus.YELLOW));
}
});
// 获取服务状态
final IndicesExistsResponse response = executeGet(new NodeCallback<IndicesExistsResponse>() {
@Override
public ActionFuture<IndicesExistsResponse> execute(final IndicesAdminClient admin) {
return admin.exists(Requests.indicesExistsRequest(indexId));
}
});
exists = response.isExists();
}
return exists;
}
/**
* 判断索引是否健康
*
* @param indexName 索引名称
* @return
*/
public boolean indexHealth(final String indexName) {
boolean health = true;
String indexId = this.getIndexId(indexName);
if (indexId != null) {
// 健康状况
final ClusterHealthResponse response = executeGet(new ClusterCallback<ClusterHealthResponse>() {
@Override
public ActionFuture<ClusterHealthResponse> execute(final ClusterAdminClient admin) {
return admin.health(Requests.clusterHealthRequest().timeout("10s"));
}
});
if (response.getIndices().get(indexId).getStatus().equals(ClusterHealthStatus.RED)) {
health = false;
}
}
return health;
}
/**
* 创建索引
*
* @param indexName 索引别名
* @param properties 索引配置
* @param force 是否强制创建索引，不论索引是否已经存在
*/
@Override
public void createIndex(final String indexName, final Map<String, String> properties, final String mappingContent, boolean force) {
String oldIndexId = this.getIndexId(indexName);
final String indexId;
if (oldIndexId == null || force) {
// 旧索引不存在 OR 旧索引已存在但强制重建
indexId = INDEX_ID_PREFIX + indexName + "_" + DateFormatUtils.format(new Date(), "yyyyMMddHHmmssSSS");
} else {
indexId = null;
}
if (indexId != null) {
// 创建新索引
executeGet(new NodeCallback<CreateIndexResponse>() {
@Override
public ActionFuture<CreateIndexResponse> execute(final IndicesAdminClient admin) {
Map<String, String> settings = new HashMap<String, String>();
settings.put("number_of_shards", properties.get("number_of_shards"));
settings.put("number_of_replicas", properties.get("number_of_replicas"));
settings.put("refresh_interval", properties.get("refresh_interval"));
settings.put("translog.flush_threshold_ops", properties.get("translog.flush_threshold_ops"));
logger.info("[client={}] [indexName={}] [indexId={}] [props={}] [mapping={}]创建索引", new Object[] { name, indexName, indexId, properties, mappingContent });
return admin.create(Requests.createIndexRequest(indexId).settings(settings).mapping(getIndexDataType(indexName), mappingContent));
}
});
if (oldIndexId != null) {
// 删除旧索引名称
this.deleteIndexName(oldIndexId, indexName);
}
// 建立别名
this.createIndexName(indexId, indexName);
if (oldIndexId != null) {
// 删除旧索引
this.deleteIndex(oldIndexId);
}
}
}
@Override
public void createIndex(final String indexName, final String tempIndexName, final Map<String, String> properties, final String mappingContent, boolean force) {
String oldIndexId = this.getIndexId(indexName);
final String indexId;
if (oldIndexId == null || force) {
// 旧索引不存在 OR 旧索引已存在但强制重建
indexId = INDEX_ID_PREFIX + tempIndexName + "_" + DateFormatUtils.format(new Date(), "yyyyMMddHHmmssSSS");
} else {
indexId = null;
}
if (indexId != null) {
// 创建新索引
executeGet(new NodeCallback<CreateIndexResponse>() {
@Override
public ActionFuture<CreateIndexResponse> execute(final IndicesAdminClient admin) {
Map<String, String> settings = new HashMap<String, String>();
settings.put("number_of_shards", properties.get("number_of_shards"));
settings.put("number_of_replicas", properties.get("number_of_replicas"));
settings.put("refresh_interval", properties.get("refresh_interval"));
settings.put("translog.flush_threshold_ops", properties.get("translog.flush_threshold_ops"));
logger.info("[client={}] [indexName={}] [indexId={}] [props={}] [mapping={}]创建索引", new Object[] { name, indexName, indexId, properties, mappingContent });
return admin.create(Requests.createIndexRequest(indexId).settings(settings).mapping(getIndexDataType(tempIndexName), mappingContent));
}
});
if (oldIndexId != null) {
// 删除旧索引名称
this.deleteIndexName(oldIndexId, tempIndexName);
}
// 建立别名
this.createIndexName(indexId, indexName);
if (oldIndexId != null) {
// 删除旧索引
this.deleteIndex(oldIndexId);
}
}
}
/**
* 删除索引
*
* @param indexName 索引名称
*/
@Override
public void deleteIndex(final String indexName) {
final String indexId = this.getIndexId(indexName);
if (indexId != null) {
executeGet(new NodeCallback<DeleteIndexResponse>() {
@Override
public ActionFuture<DeleteIndexResponse> execute(final IndicesAdminClient admin) {
return admin.delete(Requests.deleteIndexRequest(indexId));
}
});
}
}
/**
* 更新索引缓存
*
* @param indexName 索引名称
*/
@Override
public void refreshIndex(final String indexName) {
final String indexId = this.getIndexId(indexName);
if (indexId != null) {
executeGet(new NodeCallback<RefreshResponse>() {
@Override
public ActionFuture<RefreshResponse> execute(final IndicesAdminClient admin) {
return admin.refresh(Requests.refreshRequest(indexId));
}
});
}
}
/**
* 关闭索引
*
* @param indexName 索引名称
*/
@Override
public void closeIndex(final String indexName) {
final String indexId = this.getIndexId(indexName);
if (indexId != null) {
executeGet(new NodeCallback<CloseIndexResponse>() {
@Override
public ActionFuture<CloseIndexResponse> execute(final IndicesAdminClient admin) {
return admin.close(Requests.closeIndexRequest(indexId));
}
});
}
}
/**
* 刷新索引文件
*
* @param indexName 索引名称
*/
@Override
public void flushIndex(final String indexName) {
final String indexId = this.getIndexId(indexName);
if (indexId != null) {
executeGet(new NodeCallback<FlushResponse>() {
@Override
public ActionFuture<FlushResponse> execute(final IndicesAdminClient admin) {
return admin.flush(Requests.flushRequest(indexId));
}
});
}
}
/**
* 索引替换:将老索引的索引名加到新索引上，并删除老索引
*
* @param oldIndexName 旧索引名
* @param newIndexName 新索引名
*/
public void replaceIndex(final String oldIndexName, final String newIndexName) {
String oldIndexId = this.getIndexId(oldIndexName);
String newIndexId = this.getIndexId(newIndexName);
if (newIndexId != null) {
if (oldIndexId != null) {
// 1. 把老索引的索引名先删除
this.deleteIndexName(oldIndexId, oldIndexName);
}
// 2. 把新索引的索引名删除
this.deleteIndexName(newIndexId, newIndexName);
// 3. 把老索引的索引名加到新索引上面
this.createIndexName(newIndexId, oldIndexName);
if (oldIndexId != null) {
// 4. 删除老索引
this.deleteIndex(oldIndexId);
}
}
}
@Override
public void deleteTempIndex(String indexName) {
// 获取系统实际存在的索引名列表
ClusterState clusterState = client.admin().cluster().prepareState().execute().actionGet().getState();
String[] indices = clusterState.metaData().indices().keys().toArray(String.class);
ClusterStateRequest clusterStateRequest = Requests.clusterStateRequest().routingTable(true).nodes(true).indices(indices);
MetaData md = client.admin().cluster().state(clusterStateRequest).actionGet(30000).getState().getMetaData();
String regEx = "^" + INDEX_ID_PREFIX + indexName + "_(\\d){13}_(\\d){17}$";
Pattern p = Pattern.compile(regEx);
for (IndexMetaData imd : md) {
if (p.matcher(imd.index()).find()) {
boolean validate = false;
for (AliasMetaData amd : imd.aliases().values().toArray(AliasMetaData.class)) {
if (indexName.equals(amd.alias())) {
validate = true;
break;
}
}
if (!validate) {
this.deleteIndex(imd.index());
logger.info("[client={}] 删除遗留的临时索引：[id={}]", name, imd.index());
}
}
}
}
/**
* 执行索引优化
*
* @param indexName
*/
@Override
public void optimize(String indexName) {
final String indexId = this.getIndexId(indexName);
if (indexId != null) {
executeGet(new NodeCallback<OptimizeResponse>() {
@Override
public ActionFuture<OptimizeResponse> execute(final IndicesAdminClient admin) {
return admin.optimize(Requests.optimizeRequest(indexId));
}
});
}
}
/**
* 添加索引数据
*
* @param indexName 索引名称
* @param id 索引id
* @param data 索引数据
*/
@Override
public IndexResponse addIndexData(final String indexName, final String id, final Object data) {
final String indexId = this.getIndexId(indexName);
if (indexId != null) {
final IndexResponse response = executeGet(new ClientCallback<IndexResponse>() {
@Override
public ActionFuture<IndexResponse> execute(final Client client) {
String source = JSONObject.toJSONString(data);
final IndexRequest request = Requests.indexRequest(indexId).type(getIndexDataType(indexName)).id(id).source(source);
return client.index(request);
}
});
return response;
}
return null;
}
/**
* 添加索引数据
*
* @param indexName 索引名称
* @param datas 索引数据列表
*/
@Override
public BulkResponse addIndexData(final String indexName, final List<?> datas) {
if (datas != null && datas.size() > 0) {
final String indexId = this.getIndexId(indexName);
if (indexId != null) {
final BulkResponse response = executeGet(new ClientCallback<BulkResponse>() {
@Override
public ActionFuture<BulkResponse> execute(final Client client) {
final BulkRequest request = Requests.bulkRequest();
for (Object data : datas) {
request.add(Requests.indexRequest(indexId).type(getIndexDataType(indexName)).source((Map<?, ?>) data));
}
ActionFuture<BulkResponse> future = client.bulk(request);
return future;
}
});
if (response.hasFailures()) {
logger.error("[更新索引异常:{}]", response.buildFailureMessage());
}
return response;
}
}
return null;
}
/**
* 添加索引数据
*
* @param indexName 索引名称
* @param datas 索引数据列表
*/
@Override
public BulkResponse addIndexDataBean(final String indexName, final List<?> datas) {
if (datas != null && datas.size() > 0) {
final String indexId = this.getIndexId(indexName);
if (indexId != null) {
final BulkResponse response = executeGet(new ClientCallback<BulkResponse>() {
@Override
public ActionFuture<BulkResponse> execute(final Client client) {
final BulkRequest request = Requests.bulkRequest();
for (Object data : datas) {
request.add(Requests.indexRequest(indexId).type(getIndexDataType(indexName)).source(JSON.toJSONString(data)));
}
ActionFuture<BulkResponse> future = client.bulk(request);
return future;
}
});
if (response.hasFailures()) {
logger.error("[更新索引异常:{}]", response.buildFailureMessage());
}
return response;
}
}
return null;
}
/**
* 删除索引数据
*
* @param indexName
*/
@Override
public DeleteResponse deleteIndexData(final String indexName, final String id) {
final String indexId = this.getIndexId(indexName);
if (indexId != null) {
final DeleteResponse response = executeGet(new ClientCallback<DeleteResponse>() {
@Override
public ActionFuture<DeleteResponse> execute(final Client client) {
final DeleteRequest request = Requests.deleteRequest(indexId).type(getIndexDataType(indexName)).id(id);
return client.delete(request);
}
});
return response;
}
return null;
}
@Override
public DeleteByQueryResponse deleteIndexDataByQuery(final String indexName, final String query) {
final String indexId = this.getIndexId(indexName);
if (indexId != null) {
final DeleteByQueryResponse response = executeGet(new ClientCallback<DeleteByQueryResponse>() {
@Override
public ActionFuture<DeleteByQueryResponse> execute(Client client) {
final DeleteByQueryRequest request = Requests.deleteByQueryRequest(indexId).types(getIndexDataType(indexName)).source(query);
return client.deleteByQuery(request);
}
});
return response;
}
return null;
}
/**
* 更新索引数据
*
* @param indexName
*/
@Override
public UpdateResponse updateIndexData(final String indexName, final String id, final Object data) {
final String indexId = this.getIndexId(indexName);
if (indexId != null) {
final UpdateResponse response = executeGet(new ClientCallback<UpdateResponse>() {
@Override
public ActionFuture<UpdateResponse> execute(Client client) {
UpdateRequestBuilder request = new UpdateRequestBuilder(client, indexId, getIndexDataType(indexName), id);
String source = JSONObject.toJSONString(data);
request.setDoc(source).setUpsert(source);
return request.execute();
}
});
return response;
}
return null;
}
/**
* 更新索引数据
*
* @param indexName
*/
@Override
public UpdateResponse updateIndexDataStr(final String indexName, final String id, final String jsonData) {
final String indexId = this.getIndexId(indexName);
if (indexId != null) {
final UpdateResponse response = executeGet(new ClientCallback<UpdateResponse>() {
@Override
public ActionFuture<UpdateResponse> execute(Client client) {
UpdateRequestBuilder request = new UpdateRequestBuilder(client, indexId, getIndexDataType(indexName), id);
request.setDoc(jsonData).setUpsert(jsonData);
return request.execute();
}
});
return response;
}
return null;
}
@Override
public BulkResponse bulkUpdateIndexData(final String indexName, @SuppressWarnings("rawtypes") final List<Map> datas) {
final String indexId = this.getIndexId(indexName);
if (indexId != null) {
final BulkResponse response = executeGet(new ClientCallback<BulkResponse>() {
@Override
public ActionFuture<BulkResponse> execute(final Client client) {
final BulkRequest request = Requests.bulkRequest();
if (datas != null) {
for (Map<?, ?> data : datas) {
request.add(Requests.indexRequest(indexId).type(getIndexDataType(indexName)).source(data));
}
}
ActionFuture<BulkResponse> future = client.bulk(request);
return future;
}
});
if (response.hasFailures()) {
logger.error("[更新索引异常:{}]", response.buildFailureMessage());
}
return response;
}
return null;
}
// 执行集群操作请求
@SuppressWarnings("unchecked")
public <T extends ActionResponse> T executeGet(final ClusterCallback<T> callback) {
final ClusterAdminClient clusterAdmin = client.admin().cluster();
final ActionFuture<?> action = callback.execute(clusterAdmin);
final T response = (T) action.actionGet();
return response;
}
// 执行索引操作请求
@SuppressWarnings("unchecked")
public <T extends ActionResponse> T executeGet(final NodeCallback<T> callback) {
final IndicesAdminClient indicesAdmin = client.admin().indices();
final ActionFuture<?> action = callback.execute(indicesAdmin);
final T response = (T) action.actionGet();
return response;
}
/**
* 为索引创建名称
*
* @param indexId
* @param indexName
*/
public void createIndexName(String indexId, String indexName) {
// 检查别名是否存在
if (this.getIndexId(indexName) == null) {
IndicesAliasesRequest req = new IndicesAliasesRequest();
req.addAlias(indexName, indexId);
client.admin().indices().aliases(req).actionGet(30000L);
logger.info("[client=" + name + "] 创建索引[indexId=" + indexId + "]的别名[indexName=" + indexName + "]");
}
}
/**
* 为索引删除名称
*
* @param indexId
* @param indexName
*/
public void deleteIndexName(String indexId, String indexName) {
// 检查别名是否存在
if (this.getIndexId(indexName) != null) {
IndicesAliasesRequest req = new IndicesAliasesRequest();
req.removeAlias(indexId, indexName);
client.admin().indices().aliases(req).actionGet(30000L);
logger.info("[client=" + name + "] 删除索引[indexId=" + indexId + "]的别名[indexName=" + indexName + "]");
}
}
/**
* 获取索引唯一ID标识
*
* @param indexName
* @return
*/
public String getIndexId(String indexName) {
String indexId = null;
if (indexName != null) {
if (indexName.startsWith(INDEX_ID_PREFIX)) {
// 兼容输入的索引名称是索引ID的情况
indexId = indexName;
} else {
indexId = this.getIndexNameToIdMap().get(indexName);
}
}
return indexId;
}
/**
* 获取索引名称和索引ID的对应关系
*
* @return
*/
public Map<String, String> getIndexNameToIdMap() {
// 查询所有别名
Map<String, String> map = getIndexIdToNameMap();
Map<String, String> indexNameToIdMap = new HashMap<String, String>();
for (String idx : map.keySet()) {
indexNameToIdMap.put(map.get(idx), idx);
}
return indexNameToIdMap;
}
/**
* 获取索引ID和索引名称的对应关系
*
* @return
*/
public Map<String, String> getIndexIdToNameMap() {
// 获取系统实际存在的索引名列表
ClusterState clusterState = client.admin().cluster().prepareState().execute().actionGet().getState();
String[] indices = clusterState.metaData().indices().keys().toArray(String.class);
Map<String, String> indexIdToNameMap = new HashMap<String, String>();
ClusterStateRequest clusterStateRequest = Requests.clusterStateRequest().routingTable(true).nodes(true).indices(indices);
MetaData md = client.admin().cluster().state(clusterStateRequest).actionGet(30000).getState().metaData();
for (IndexMetaData imd : md) {
for (AliasMetaData amd : imd.aliases().values().toArray(AliasMetaData.class)) {
indexIdToNameMap.put(imd.index(), amd.alias());
}
}
return indexIdToNameMap;
}
/**
* 通过索引名称获取索引数据类型 标准索引名称中不含下划线，临时索引名称含义下划线，以下划线分割，前半部为标准索引名称，后半部为时间戳 参考方法
* generateTempIndexName 在本系统中，索引数据类型与标准索引名称保持一致，因此如果为临时索引名称，取前半部
*
* @param indexName
* @return
*/
public String getIndexDataType(String indexName) {
return indexName.split("_")[0];
}
/**
* 根据indexName找到所有相关的
*
* @return
*/
public List<String> getRelatedIndexs(String name) {
List<String> result = new ArrayList<String>();
Map<String, String> idToNameMap = getIndexIdToNameMap();
Set<String> nameSet = idToNameMap.keySet();
String regEx = "^" + INDEX_ID_PREFIX + name + "_(\\d){13}_(\\d){17}$";
Pattern p = Pattern.compile(regEx);
for (String indexId : nameSet) {
if (p.matcher(indexId).find()) {
result.add(idToNameMap.get(indexId));
}
}
return result;
}
@Override
public void setIndexSettings(String indexName, Map<String, String> settings) {
String indexId = this.getIndexId(indexName);
if (indexId != null && settings != null && settings.size() > 0) {
ImmutableSettings.Builder updateSettings = ImmutableSettings.settingsBuilder();
Set<String> keySet = settings.keySet();
for (String key : keySet) {
updateSettings.put(key, settings.get(key));
}
client.admin().indices().prepareUpdateSettings(indexId).setSettings(updateSettings).execute().actionGet();
}
}
@Override
public Map<String, String> optimumSettingsForBulkIndexing(String indexName) {
String indexId = this.getIndexId(indexName);
IndexMetaData indexMetaData = client.admin().cluster().prepareState().execute().actionGet().getState().metaData().index(indexId);
Map<String, String> map = new HashMap<String, String>();
map.put("index.refresh_interval", indexMetaData.settings().get("index.refresh_interval"));
map.put("index.number_of_replicas", indexMetaData.settings().get("index.number_of_replicas"));
map.put("index.translog.flush_threshold_ops", indexMetaData.settings().get("index.translog.flush_threshold_ops"));
ImmutableSettings.Builder updateSettings = ImmutableSettings.settingsBuilder();
updateSettings.put("index.refresh_interval", -1);
updateSettings.put("index.number_of_replicas", 0);
updateSettings.put("index.translog.flush_threshold_ops", 100000);
client.admin().indices().prepareUpdateSettings(indexId).setSettings(updateSettings).execute().actionGet();
return map;
}
}
... ...
/**
* Project: petstore-dal
*
* File Created at 2015年11月4日
* $Id$
*
* Copyright 2015 Yoho.cn Croporation Limited.
* All rights reserved.
*
* This software is the confidential and proprietary information of
* Yoho Company. ("Confidential Information"). You shall not
* disclose such Confidential Information and shall use it only in
* accordance with the terms of the license agreement you entered into
* with Yoho.cn.
*/
package com.taobao.sample.petstore.dal.elasticsearch.client.impl;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.commons.lang.StringUtils;
import org.elasticsearch.action.ActionFuture;
import org.elasticsearch.action.ActionResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.get.MultiGetItemResponse;
import org.elasticsearch.action.get.MultiGetRequest;
import org.elasticsearch.action.get.MultiGetResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.Requests;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHitField;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.AbstractAggregationBuilder;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.highlight.HighlightField;
import org.elasticsearch.search.sort.SortOrder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.taobao.sample.petstore.dal.elasticsearch.callback.ClientCallback;
import com.taobao.sample.petstore.dal.elasticsearch.client.ElasticClientTemplate;
import com.taobao.sample.petstore.dal.elasticsearch.entity.ElasticQuery;
import com.taobao.sample.petstore.dal.elasticsearch.entity.ElasticResult;
/**
* TODO Comment of ElasticClientImpl
*
* @author ibm
* @version $Id: ElasticClientImpl.java 2015年11月4日 上午1:40:28 $
*/
public class ElasticClientTemplateImpl implements ElasticClientTemplate {
public static final String INDEX_ID_PREFIX = "yoho_";
protected final Logger logger = LoggerFactory.getLogger(ElasticClientTemplateImpl.class);
protected Client client;
protected String name;
public ElasticClientTemplateImpl() {
}
public ElasticClientTemplateImpl(Client client) {
this.client = client;
}
/**
* 获取索引器名称
*
* @return
*/
@Override
public String getName() {
return this.name;
}
/**
* 设置索引器名称
*
* @param name
*/
@Override
public void setName(String name) {
this.name = name;
}
/**
* 通过关键词搜索数据
*
* @param indexName
* @param elasticQuery
* @return ElasticResult
*/
@Override
public ElasticResult search(final String[] indexName, final ElasticQuery elasticQuery) {
long begin = System.currentTimeMillis();
logger.debug("[model=ElasticClientTemplate][method=search][indexNameList={}][begin={}]", indexName, begin);
//////////////////////1、组装查询参数
//////////////////////2、执行查询操作
final SearchResponse response = executeGet(new ClientCallback<SearchResponse>() {
@Override
public ActionFuture<SearchResponse> execute(final Client client) {
final SearchSourceBuilder sourceBuilder = SearchSourceBuilder.searchSource();
// 搜索条件
if (elasticQuery.getFiter() != null) {
sourceBuilder.query(QueryBuilders.filteredQuery(elasticQuery.getQuery(), elasticQuery.getFiter()));
} else {
sourceBuilder.query(elasticQuery.getQuery());
}
// 每页记录数
sourceBuilder.size(elasticQuery.getSize());
// 当前页起始记录
sourceBuilder.from(elasticQuery.getOffset());
// 显示字段，默认全部显示
if (elasticQuery.getResultFields().size() > 0) {
sourceBuilder.fields(elasticQuery.getResultFields());
}
if (elasticQuery.isHighlight()) {
for (String field : elasticQuery.getHighlightFields()) {
sourceBuilder.highlighter().field(field).preTags(elasticQuery.getHighlightPreTag()).postTags(elasticQuery.getHighlightPostTag());
}
}
// 分组统计
List<AbstractAggregationBuilder> aggregations = elasticQuery.getAggregationBuilders();
if (aggregations != null && aggregations.size() > 0) {
for (AbstractAggregationBuilder aggregation : aggregations) {
sourceBuilder.aggregation(aggregation);
}
}
// 排序字段
for (String sortField : elasticQuery.getSortFields().keySet()) {
String sortValue = elasticQuery.getSortFields().get(sortField);
sourceBuilder.sort(sortField, SortOrder.ASC.toString().equals(sortValue) ? SortOrder.ASC : SortOrder.DESC);
}
SearchRequest request = Requests.searchRequest(indexName);
request.source(sourceBuilder.toString());
if (elasticQuery.getSearchType() != null) {
request.searchType(elasticQuery.getSearchType());
}
logger.trace("[client={}] [query={}]", name, sourceBuilder.toString());
return client.search(request);
}
});
///////////构造查询返回结果
ElasticResult ElasticResult = new ElasticResult();
SearchHits hits = response.getHits();
List<Map<String, Object>> resultList = new ArrayList<Map<String, Object>>();
for (SearchHit hit : hits.getHits()) {
Map<String, Object> map;
if (hit.getSource() != null) {
map = hit.getSource();
} else {
map = new HashMap<String, Object>();
}
if (hit.getFields() != null && hit.getFields().size() > 0) {
// 返回指定字段
for (SearchHitField shf : hit.fields().values()) {
map.put(shf.name(), shf.value());
}
}
if (elasticQuery.isHighlight()) {
// 添加高亮字段信息
Map<String, Object> highlightMap = new HashMap<String, Object>();
// highlightMap.putAll(map);
for (HighlightField hf : hit.highlightFields().values()) {
highlightMap.put(hf.name(), hf.fragments()[0].toString());
}
map.put("_highlight", highlightMap);
}
resultList.add(map);
}
ElasticResult.setResultList(resultList);
ElasticResult.setTotal(hits.getTotalHits());
ElasticResult.setPage(elasticQuery.getPage());
if (ElasticResult.getTotal() % elasticQuery.getSize() == 0) {
ElasticResult.setTotalPage(ElasticResult.getTotal() / elasticQuery.getSize());
} else {
ElasticResult.setTotalPage(ElasticResult.getTotal() / elasticQuery.getSize() + 1);
}
if (response.getAggregations() != null) {
Map<String, Aggregation> aggregations = response.getAggregations().asMap();
ElasticResult.setAggMaps(aggregations);
}
logger.debug("[model=EsIndexClient][method=search][indexNameList={}][cost={}ms]", indexName, System.currentTimeMillis() - begin);
return ElasticResult;
}
/**
* 通过关键词搜索数据
*
* @param indexName
* @param elasticQuery
* @return ElasticResult
*/
@Override
public ElasticResult search(final String indexName, final ElasticQuery ElasticQuery) {
if (StringUtils.isBlank(indexName))
return null;
return search(indexName, ElasticQuery);
}
/**
* 通过id检索数据
*
* @param indexName
* @param elasticQuery
* @return ElasticResult
*/
@Override
public List<Map<String, Object>> multiGet(final String indexName, final Set<String> idList, final List<String> fields) {
final MultiGetResponse response = executeGet(new ClientCallback<MultiGetResponse>() {
@Override
public ActionFuture<MultiGetResponse> execute(final Client client) {
MultiGetRequest request = new MultiGetRequest();
for (String id : idList) {
MultiGetRequest.Item item = new MultiGetRequest.Item(indexName, null, id);
if (fields != null && fields.size() > 0) {
item.fields((String[]) fields.toArray(new String[fields.size()]));
}
request.add(item);
}
return client.multiGet(request);
}
});
return makeResponse(response, indexName);
}
/**
* 通过id检索数据
*
* @param indexName
* @param elasticQuery
* @return ElasticResult
*/
@Override
public List<Map<String, Object>> multiGet(final String indexName, final String[] idList, final List<String> fields) {
final MultiGetResponse response = executeGet(new ClientCallback<MultiGetResponse>() {
@Override
public ActionFuture<MultiGetResponse> execute(final Client client) {
MultiGetRequest request = new MultiGetRequest();
for (String id : idList) {
MultiGetRequest.Item item = new MultiGetRequest.Item(indexName, null, id);
if (fields != null && fields.size() > 0) {
item.fields((String[]) fields.toArray(new String[fields.size()]));
}
request.add(item);
}
return client.multiGet(request);
}
});
return makeResponse(response, indexName);
}
/**
* 通过id检索数据
*
* @param indexName
* @param elasticQuery
* @return ElasticResult
*/
@Override
public Map<String, Object> get(final String indexName, final String id) {
final GetResponse response = executeGet(new ClientCallback<GetResponse>() {
@Override
public ActionFuture<GetResponse> execute(final Client client) {
GetRequest request = new GetRequest(indexName, indexName, id);
return client.get(request);
}
});
// 构造返回结果
if (!response.getSource().isEmpty()) {
return response.getSource();
}
return null;
}
private List<Map<String, Object>> makeResponse(final MultiGetResponse response, final String indexName) {
// 构造返回结果
List<Map<String, Object>> list = new ArrayList<Map<String, Object>>();
Map<String, Object> tmpMap;
for (MultiGetItemResponse item : response.getResponses()) {
if (item.getResponse().isExists()) {
tmpMap = item.getResponse().getSource();
list.add(tmpMap);
}
}
return list;
}
/**
* 执行数据操作请求
*
* @param callback
* @return
*/
protected <T extends ActionResponse> T executeGet(final ClientCallback<T> callback) {
final ActionFuture<T> action = callback.execute(client);
final T response = action.actionGet();
return response;
}
}
... ...
/**
* Project: petstore-dal
*
* File Created at 2015年11月4日
* $Id$
*
* Copyright 2015 Yoho.cn Croporation Limited.
* All rights reserved.
*
* This software is the confidential and proprietary information of
* Yoho Company. ("Confidential Information"). You shall not
* disclose such Confidential Information and shall use it only in
* accordance with the terms of the license agreement you entered into
* with Yoho.cn.
*/
package com.taobao.sample.petstore.dal.elasticsearch.entity;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.index.query.FilterBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.search.aggregations.AbstractAggregationBuilder;
/**
* 封装Elasticsearch查询参数
*
* @author ibm
* @version $Id: ElasticsearchParam.java 2015年11月4日 上午1:26:25 $
*/
public class ElasticQuery implements Serializable, Cloneable {
private static final long serialVersionUID = 1L;
/* 排序字段 ,有先后顺序 */
private LinkedHashMap<String, String> sortFields = new LinkedHashMap<String, String>();
/* 查询框参数 */
private QueryBuilder query;
/* 过滤参数 */
private FilterBuilder fiter;
/* 开始位置 */
private int offset = 0;
/* 页码 */
private int page = 1;
/* 要返回条数 */
private int size = 1;
/* 返回字段 */
private List<String> resultFields = new ArrayList<String>();
/* 是否高亮显示 */
private boolean highlight = false;
/* 是否高亮显示 */
private String highlightPreTag = "<span style=\"color: #CC0000;\">";
/* 是否高亮显示 */
private String highlightPostTag = "</span>";
private SearchType searchType = null;//如果不需要返回列表的时候可以将SearchType设为count，减少返回
/* 搜索结果字段 默认显示所有存储的字段 */
private List<String> highlightFields = new ArrayList<String>();
private List<AbstractAggregationBuilder> aggregationBuilders;
public LinkedHashMap<String, String> getSortFields() {
return sortFields;
}
public void setSortFields(LinkedHashMap<String, String> sortFields) {
this.sortFields = sortFields;
}
public QueryBuilder getQuery() {
return query;
}
public void setQuery(QueryBuilder query) {
this.query = query;
}
public FilterBuilder getFiter() {
return fiter;
}
public void setFiter(FilterBuilder fiter) {
this.fiter = fiter;
}
public int getOffset() {
return offset;
}
public void setOffset(int offset) {
this.offset = offset;
}
public int getPage() {
return page;
}
public void setPage(int page) {
this.page = page;
}
public int getSize() {
return size == 0 ? 60 : size;
}
public void setSize(int size) {
this.size = size == 0 ? 60 : size;
}
public List<String> getResultFields() {
return resultFields;
}
public void setResultFields(List<String> resultFields) {
this.resultFields = resultFields;
}
public boolean isHighlight() {
return highlight;
}
public void setHighlight(boolean highlight) {
this.highlight = highlight;
}
public String getHighlightPreTag() {
return highlightPreTag;
}
public void setHighlightPreTag(String highlightPreTag) {
this.highlightPreTag = highlightPreTag;
}
public String getHighlightPostTag() {
return highlightPostTag;
}
public void setHighlightPostTag(String highlightPostTag) {
this.highlightPostTag = highlightPostTag;
}
public List<String> getHighlightFields() {
return highlightFields;
}
public void setHighlightFields(List<String> highlightFields) {
this.highlightFields = highlightFields;
}
public List<AbstractAggregationBuilder> getAggregationBuilders() {
return aggregationBuilders;
}
public void setAggregationBuilders(List<AbstractAggregationBuilder> aggregationBuilders) {
this.aggregationBuilders = aggregationBuilders;
}
public SearchType getSearchType() {
return searchType;
}
public void setSearchType(SearchType searchType) {
this.searchType = searchType;
}
@Override
public ElasticQuery clone() {
ElasticQuery param = null;
try {
param = (ElasticQuery) super.clone();
} catch (CloneNotSupportedException e) {
e.printStackTrace();
}
return param;
}
}
... ...
/**
* Project: petstore-dal
*
* File Created at 2015年11月4日
* $Id$
*
* Copyright 2015 Yoho.cn Croporation Limited.
* All rights reserved.
*
* This software is the confidential and proprietary information of
* Yoho Company. ("Confidential Information"). You shall not
* disclose such Confidential Information and shall use it only in
* accordance with the terms of the license agreement you entered into
* with Yoho.cn.
*/
package com.taobao.sample.petstore.dal.elasticsearch.entity;
import java.io.Serializable;
import java.util.List;
import java.util.Map;
import org.elasticsearch.search.aggregations.Aggregation;
/**
* 封装Elasticsearch搜索返回结果
*
* @author ibm
* @version $Id: ElasticsearchResult.java 2015年11月4日 上午1:26:43 $
*/
public class ElasticResult implements Serializable {
private static final long serialVersionUID = 1L;
/* 结果 */
private List<Map<String, Object>> resultList;
private long total;
private int page;
private long totalPage;
private Map<String, Aggregation> aggMaps;
public List<Map<String, Object>> getResultList() {
return resultList;
}
public void setResultList(List<Map<String, Object>> resultList) {
this.resultList = resultList;
}
public long getTotal() {
return total;
}
public void setTotal(long total) {
this.total = total;
}
public Map<String, Aggregation> getAggMaps() {
return aggMaps;
}
public void setAggMaps(Map<String, Aggregation> aggMaps) {
this.aggMaps = aggMaps;
}
public int getPage() {
return page;
}
public void setPage(int page) {
this.page = page;
}
public long getTotalPage() {
return totalPage;
}
public void setTotalPage(long totalPage) {
this.totalPage = totalPage;
}
}
... ...
/**
* Project: petstore-dal
*
* File Created at 2015年11月3日
* $Id$
*
* Copyright 2015 Yoho.cn Croporation Limited.
* All rights reserved.
*
* This software is the confidential and proprietary information of
* Yoho Company. ("Confidential Information"). You shall not
* disclose such Confidential Information and shall use it only in
* accordance with the terms of the license agreement you entered into
* with Yoho.cn.
*/
package com.taobao.sample.petstore.dal.elasticsearch.factory;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.ImmutableSettings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import com.taobao.sample.petstore.dal.elasticsearch.client.impl.ElasticAdminClientTemplateImpl;
/**
* TODO Comment of ElasticsearchClientFactory
*
* @author ibm
* @version $Id: ElasticsearchClientFactory.java 2015年11月3日 下午11:49:17 $
*/
public class ElasticClientFactory implements FactoryBean<ElasticAdminClientTemplateImpl>, InitializingBean, DisposableBean {
protected final Logger logger = LoggerFactory.getLogger(ElasticClientFactory.class);
private ElasticConfig config;
private Client client;
private ElasticAdminClientTemplateImpl elasticAdminClient;
/**
* @param config the config to set
*/
public void setConfig(ElasticConfig config) {
this.config = config;
}
@Override
public ElasticAdminClientTemplateImpl getObject() throws Exception {
return this.elasticAdminClient;
}
@Override
public Class<?> getObjectType() {
return this.elasticAdminClient == null ? ElasticAdminClientTemplateImpl.class : this.elasticAdminClient.getClass();
}
@Override
public boolean isSingleton() {
return true;
}
@Override
public void destroy() throws Exception {
client.close();
}
@Override
public void afterPropertiesSet() throws Exception {
this.client = initClient();
this.elasticAdminClient = new ElasticAdminClientTemplateImpl(this.client);
}
private Client initClient() {
String clusterName = config.getClusterName();
String[] serverHosts = config.getServerHosts();
String pingTimeout = config.getPingTimeout();
Settings settings = ImmutableSettings.settingsBuilder().put("cluster.name", clusterName).put("client.transport.ping_timeout", pingTimeout).build();
TransportClient transportClient = new TransportClient(settings);
for (String serverHost : serverHosts) {
String[] servers = serverHost.split(":");
String ipAddress = servers[0];
int port = 9300;
if (servers.length > 1) {
port = Integer.parseInt(servers[1]);
}
transportClient.addTransportAddress(new InetSocketTransportAddress(ipAddress, port));
}
logger.info("Setup transport client {}, cluster name {}", serverHosts, clusterName);
return transportClient;
}
}
... ...
/**
* Project: petstore-dal
*
* File Created at 2015年11月3日
* $Id$
*
* Copyright 2015 Yoho.cn Croporation Limited.
* All rights reserved.
*
* This software is the confidential and proprietary information of
* Yoho Company. ("Confidential Information"). You shall not
* disclose such Confidential Information and shall use it only in
* accordance with the terms of the license agreement you entered into
* with Yoho.cn.
*/
package com.taobao.sample.petstore.dal.elasticsearch.factory;
import org.apache.commons.lang.builder.ToStringBuilder;
/**
* TODO Comment of ElasticsearchConfig
*
* @author ibm
* @version $Id: ElasticsearchConfig.java 2015年11月3日 下午11:55:52 $
*/
public class ElasticConfig {
private String clusterName;
private String[] serverHosts;
private String pingTimeout;
/**
* @return the clusterName
*/
public String getClusterName() {
return clusterName;
}
/**
* @param clusterName the clusterName to set
*/
public void setClusterName(String clusterName) {
this.clusterName = clusterName;
}
/**
* @return the serverHosts
*/
public String[] getServerHosts() {
return serverHosts;
}
/**
* @param serverHosts the serverHosts to set
*/
public void setServerHosts(String[] serverHosts) {
this.serverHosts = serverHosts;
}
/**
* @return the pingTimeout
*/
public String getPingTimeout() {
return pingTimeout;
}
/**
* @param pingTimeout the pingTimeout to set
*/
public void setPingTimeout(String pingTimeout) {
this.pingTimeout = pingTimeout;
}
public String toString() {
return new ToStringBuilder(this).append("clusterName", clusterName).append("serverHosts", serverHosts).append("pingTimeout", pingTimeout).toString();
}
}
... ...
... ... @@ -11,6 +11,15 @@
<bean id="genericDao" class="com.taobao.sample.petstore.dal.dao.ibatis.IbatisGenericDaoImpl">
<property name="sqlSessionTemplate" ref="sqlSessionTemplate"></property>
<property name="sqlSessionFactory" ref="sqlSessionFactory"></property>
<property name="sqlTimeLimits" value="5"></property>
<property name="showSql" value="true"></property>
</bean>
<bean id="mybatisGenericDao" class="com.taobao.sample.petstore.dal.dao.mybatis.MybatisGenericDaoImpl">
<property name="sqlSessionTemplate" ref="sqlSessionTemplate"></property>
<property name="sqlSessionFactory" ref="sqlSessionFactory"></property>
<property name="sqlTimeLimits" value="5"></property>
<property name="showSql" value="true"></property>
</bean>
<bean id="genericJdbcDao" class="com.taobao.sample.petstore.dal.dao.jdbc.JdbcGenericDaoImpl">
... ...