从JDBC ResultSet创建对象流

Java 8中引入了Stream API和Lambda功能，使我们能够从JDBC ResultSet优雅地转换为仅提供映射功能的对象流。这种功能当然可以是lambda。基本上，这个想法是使用ResultSet作为Supplier来生成Stream：

public class ResultSetSupplier implements Supplier<T>{private final ResultSet rs;private final Function<ResultSet, T> mappingFunction;private ResultSetSupplier(ResultSet rs,Function<ResultSet, T> mappingFunction) {this.rs = rs;this.mappingFunction = mappingFunction;}@Overridepublic T get() {try {if (rs.next())return mappingFunction.apply(rs);} catch (SQLException e) {e.printStackTrace();}return null;}}

参数mappingFunction可能是lambda表达式，用于从ResultSet构建T实例。就像ActiveRecord模式一样，此类ResultSet中的每一行都映射到T的实例，其中列是T的属性。让我们考虑类City ：

public class City{String city;String country;public City(String city, String country) {this.city = city;this.country = country;}public String getCountry() {return country;}@Overridepublic String toString() {return "City [city=" + city + ", country=" + country + ";]";}@Overridepublic int hashCode() {final int prime = 31;int result = 1;result = prime * result + ((city == null) ? 0 : city.hashCode());result = prime * result+ ((country == null) ? 0 : country.hashCode());return result;}@Overridepublic boolean equals(Object obj) {if (this == obj)return true;if (obj == null)return false;if (getClass() != obj.getClass())return false;City other = (City) obj;if (city == null) {if (other.city != null)return false;} else if (!city.equals(other.city))return false;if (country == null) {if (other.country != null)return false;} else if (!country.equals(other.country))return false;return true;}}

City对象的映射函数可以是lambda表达式，如下所示：

(ResultSet rs) -> {try{return new City(rs.getString("city"), rs.getString("country"));} catch (Exception e) {return null;}}

我们假设数据库列分别称为city和country 。尽管PreparedStatement和ResultSet都实现了AutoCloseable接口，但必须提供resultSet才能创建对象流，但在关闭流时也必须关闭此类resultSet。一种可能的方法是使用代理来拦截对象流上的方法调用。因此，当在代理上调用close（）方法时，它将在提供的resultSet上调用close（） 。为了能够提供所有Stream功能，所有方法调用也将在对象流上被调用。使用代理很容易实现。我们来看一下。我们将有一个代理工厂和一个调用处理程序：

public class ResultSetStreamInvocationHandler<T> implements InvocationHandler{private Stream<T> stream; // proxy will intercept method calls to such streamprivate PreparedStatement st;private ResultSet rs;public void setup(PreparedStatement st, Function<ResultSet, T> mappingFunction)throws SQLException{// PreparedStatement must be already setup in order// to just call executeQuery()this.st = st;rs = st.executeQuery();stream = Stream.generate(new ResultSetSupplier(rs, mappingFunction));}@Overridepublic Object invoke(Object proxy, Method method, Object[] args)throws Throwable {if (method == null)throw new RuntimeException("null method null");// implement AutoCloseable for PreparedStatement// as calling close() more than once has no effectsif (method.getName().equals("close") && args == null){// invoked close(), no argumentsif (st != null){st.close(); // closes ResultSet too}}return method.invoke(stream, args);}private class ResultSetSupplier implements Supplier<T>{private final ResultSet rs;private final Function<ResultSet, T> mappingFunction;private ResultSetSupplier(ResultSet rs, Function<ResultSet, T> mappingFunction) {this.rs = rs;this.mappingFunction = mappingFunction;}@Overridepublic T get() {try {if (rs.next())return mappingFunction.apply(rs);} catch (SQLException e) {e.printStackTrace();}return null;}
}}

请注意如何使用invoke来拦截方法调用。在接近的情况下（）被调用时， 关闭（）调用的PreparedStatement为好。对于每个调用的方法，将在代理的流中调用相应的方法调用。和工厂：

public class ResultSetStream<T>{@SuppressWarnings("unchecked")public Stream<T> getStream(PreparedStatement st,Function<ResultSet, T> mappingFunction) throws SQLException{final ResultSetStreamInvocationHandler<T> handler =new ResultSetStreamInvocationHandler<T>();handler.setup(st, mappingFunction);Stream<T> proxy = (Stream<T>) Proxy.newProxyInstance(getClass().getClassLoader(),new Class<?>[] {Stream.class},handler);return proxy;}
}

综上所述，让我们编写一个简单的测试来显示用法。 Mockito将用于模拟PreparedStatement和ResultSet以避免对实际数据库运行测试。

public class ResultSetStreamTest {private class City{String city;String country;public City(String city, String country) {this.city = city;this.country = country;}public String getCountry() {return country;}@Overridepublic String toString() {return "City [city=" + city + ", country=" + country + "]";}@Overridepublic int hashCode() {final int prime = 31;int result = 1;result = prime * result + getOuterType().hashCode();result = prime * result + ((city == null) ? 0 : city.hashCode());result = prime * result+ ((country == null) ? 0 : country.hashCode());return result;}@Overridepublic boolean equals(Object obj) {if (this == obj)return true;if (obj == null)return false;if (getClass() != obj.getClass())return false;City other = (City) obj;if (!getOuterType().equals(other.getOuterType()))return false;if (city == null) {if (other.city != null)return false;} else if (!city.equals(other.city))return false;if (country == null) {if (other.country != null)return false;} else if (!country.equals(other.country))return false;return true;}private ResultSetStreamTest getOuterType() {return ResultSetStreamTest.this;}}private String[][] data = new String[][]{{"Karachi", "Pakistan"},{"Istanbul", "Turkey"},{"Hong Kong", "China"},{"Saint Petersburg", "Russia"},{"Sydney", "Australia"},{"Berlin", "Germany"},{"Madrid", "Spain"}};private int timesCalled;private PreparedStatement mockPST;private ResultSet mockRS;@Beforepublic void setup() throws SQLException{timesCalled = -1;mockRS = mock(ResultSet.class);mockPST = mock(PreparedStatement.class);when(mockRS.next()).thenAnswer(new Answer<Boolean>() {@Overridepublic Boolean answer(InvocationOnMock invocation) throws Throwable {if (timesCalled++ >= data.length)return false;return true;}});when(mockRS.getString(eq("city"))).thenAnswer(new Answer<String>() {@Overridepublic String answer(InvocationOnMock invocation) throws Throwable {return data[timesCalled][0];}});when(mockRS.getString(eq("country"))).thenAnswer(new Answer<String>() {@Overridepublic String answer(InvocationOnMock invocation) throws Throwable {return data[timesCalled][1];}});when(mockPST.executeQuery()).thenReturn(mockRS);}@Testpublic void simpleTest() throws SQLException{try (Stream<City> testStream = new ResultSetStream<City>().getStream(mockPST,(ResultSet rs) -> {try {return new City(rs.getString("city"), rs.getString("country"));} catch (Exception e) {return null;}})){Iterator<City> cities = testStream.filter(city -> !city.getCountry().equalsIgnoreCase("China")).limit(3).iterator();assertTrue(cities.hasNext());assertEquals(new City("Karachi", "Pakistan"), cities.next());assertTrue(cities.hasNext());assertEquals(new City("Istanbul", "Turkey"), cities.next());assertTrue(cities.hasNext());assertEquals(new City("Saint Petersburg", "Russia"), cities.next());assertFalse(cities.hasNext());}}}